import pymysql
import os
import threading
import hashlib
import time
import json
import random
import string
import aiohttp
import io
import base64
from datetime import datetime
from PIL import Image, ImageDraw, ImageFont
from typing import List, Tuple, Dict, Optional, Any
from loguru import logger

class DBManager:
    """MySQL数据库管理，持久化存储Cookie和关键字"""
    
    def __init__(self, db_config: dict = None):
        """初始化数据库连接和表结构"""
        # 默认MySQL配置
        if db_config is None:
            db_config = {
                'host': 'home.jhsrvip.cn',
                'user': 'root', 
                'password': '123589',
                'database': 'xianyu_data',
                'port': 3306,
                'charset': 'utf8mb4',
                'autocommit': True
            }

        # 支持环境变量覆盖配置
        self.db_config = {
            'host': os.getenv('MYSQL_HOST', db_config.get('host', 'home.jhsrvip.cn')),
            'user': os.getenv('MYSQL_USER', db_config.get('user', 'root')),
            'password': os.getenv('MYSQL_PASSWORD', db_config.get('password', '123589')),
            'database': os.getenv('MYSQL_DATABASE', db_config.get('database', 'xianyu_data')),
            'port': int(os.getenv('MYSQL_PORT', db_config.get('port', 3306))),
            'charset': db_config.get('charset', 'utf8mb4'),
            'autocommit': db_config.get('autocommit', True)
        }
        
        logger.info(f"MySQL数据库配置: {self.db_config['host']}:{self.db_config['port']}/{self.db_config['database']}")
        self.conn = None
        self.lock = threading.RLock()  # 使用可重入锁保护数据库操作

        # SQL日志配置 - 默认启用
        self.sql_log_enabled = True  # 默认启用SQL日志
        self.sql_log_level = 'INFO'  # 默认使用INFO级别

        # 允许通过环境变量覆盖默认设置
        if os.getenv('SQL_LOG_ENABLED'):
            self.sql_log_enabled = os.getenv('SQL_LOG_ENABLED', 'true').lower() == 'true'
        if os.getenv('SQL_LOG_LEVEL'):
            self.sql_log_level = os.getenv('SQL_LOG_LEVEL', 'INFO').upper()

        logger.info(f"SQL日志已启用，日志级别: {self.sql_log_level}")

        # 建立数据库连接
        self.conn = self._create_connection()
        self.init_db()
    
    def _create_connection(self):
        """创建MySQL数据库连接"""
        try:
            conn = pymysql.connect(**self.db_config)
            logger.info("MySQL数据库连接成功")
            return conn
        except Exception as e:
            logger.error(f"MySQL数据库连接失败: {e}")
            raise 
    
    def init_db(self):
        """初始化数据库"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 创建cookies表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS cookies (
                    id VARCHAR(255) PRIMARY KEY,
                    value TEXT NOT NULL,
                    user_id INT,
                    remark TEXT,
                    auto_confirm BOOLEAN DEFAULT FALSE,
                    pause_duration INT DEFAULT 10,
                    pause_enabled BOOLEAN DEFAULT TRUE,
                    token_status VARCHAR(50) DEFAULT 'unknown',
                    token_failed_at TIMESTAMP NULL,
                    notification_wxid TEXT,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建keywords表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS keywords (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    cookie_id VARCHAR(255) NOT NULL,
                    keyword TEXT NOT NULL,
                    reply TEXT NOT NULL,
                    item_id VARCHAR(255),
                    type VARCHAR(50) DEFAULT 'text',
                    image_url TEXT,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    FOREIGN KEY (cookie_id) REFERENCES cookies(id) ON DELETE CASCADE,
                    UNIQUE KEY unique_keyword (cookie_id, keyword(191), item_id)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建notification_channels表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS notification_channels (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    name VARCHAR(255) NOT NULL,
                    channel_type VARCHAR(50) NOT NULL,
                    config TEXT NOT NULL,
                    enabled BOOLEAN DEFAULT TRUE,
                    user_id INT,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建message_notifications表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS message_notifications (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    cookie_id VARCHAR(255) NOT NULL,
                    channel_id INT NOT NULL,
                    enabled BOOLEAN DEFAULT TRUE,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    FOREIGN KEY (cookie_id) REFERENCES cookies(id) ON DELETE CASCADE,
                    FOREIGN KEY (channel_id) REFERENCES notification_channels(id) ON DELETE CASCADE,
                    UNIQUE KEY unique_notification (cookie_id, channel_id)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建system_settings表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS system_settings (
                    `key` VARCHAR(255) PRIMARY KEY,
                    value TEXT,
                    description TEXT,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建permanently_paused_chats表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS permanently_paused_chats (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    chat_id VARCHAR(255) NOT NULL UNIQUE,
                    cookie_id VARCHAR(255) NOT NULL,
                    reason VARCHAR(255) DEFAULT '转人工',
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    FOREIGN KEY (cookie_id) REFERENCES cookies(id) ON DELETE CASCADE
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建users表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS users (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    username VARCHAR(255) UNIQUE NOT NULL,
                    email VARCHAR(255) UNIQUE NOT NULL,
                    password_hash VARCHAR(255) NOT NULL,
                    phone VARCHAR(20),
                    nickname VARCHAR(255),
                    is_active BOOLEAN DEFAULT TRUE,
                    is_admin BOOLEAN DEFAULT FALSE,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')
                
                # 添加phone和nickname字段（如果不存在）
                try:
                    cursor.execute('ALTER TABLE users ADD COLUMN phone VARCHAR(20)')
                except Exception:
                    pass  # 字段已存在
                
                try:
                    cursor.execute('ALTER TABLE users ADD COLUMN nickname VARCHAR(255)')
                except Exception:
                    pass  # 字段已存在

                # 创建角色表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS roles (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    name VARCHAR(255) UNIQUE NOT NULL,
                    description TEXT,
                    is_system BOOLEAN DEFAULT FALSE COMMENT '系统角色不可删除',
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建系统权限表（避免与现有permissions表冲突）
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS system_permissions (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    name VARCHAR(255) UNIQUE NOT NULL,
                    description TEXT,
                    category VARCHAR(100) DEFAULT 'general',
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建角色权限关联表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS role_permissions (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    role_id INT NOT NULL,
                    permission_id INT NOT NULL,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    FOREIGN KEY (role_id) REFERENCES roles(id) ON DELETE CASCADE,
                    FOREIGN KEY (permission_id) REFERENCES system_permissions(id) ON DELETE CASCADE,
                    UNIQUE KEY unique_role_permission (role_id, permission_id)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建用户角色关联表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS user_roles (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    user_id INT NOT NULL,
                    role_id INT NOT NULL,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
                    FOREIGN KEY (role_id) REFERENCES roles(id) ON DELETE CASCADE,
                    UNIQUE KEY unique_user_role (user_id, role_id)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建菜单表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS menus (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    name VARCHAR(255) NOT NULL,
                    menu_key VARCHAR(255) UNIQUE NOT NULL COMMENT '菜单标识符',
                    icon VARCHAR(100),
                    url VARCHAR(255),
                    parent_id INT DEFAULT NULL,
                    sort_order INT DEFAULT 0,
                    is_system BOOLEAN DEFAULT FALSE COMMENT '系统菜单不可删除',
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    FOREIGN KEY (parent_id) REFERENCES menus(id) ON DELETE SET NULL
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建角色菜单关联表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS role_menus (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    role_id INT NOT NULL,
                    menu_id INT NOT NULL,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    FOREIGN KEY (role_id) REFERENCES roles(id) ON DELETE CASCADE,
                    FOREIGN KEY (menu_id) REFERENCES menus(id) ON DELETE CASCADE,
                    UNIQUE KEY unique_role_menu (role_id, menu_id)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建verification_codes表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS verification_codes (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    email VARCHAR(255) NOT NULL,
                    code VARCHAR(10) NOT NULL,
                    code_type VARCHAR(50) NOT NULL,
                    expires_at TIMESTAMP NOT NULL,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建email_verifications表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS email_verifications (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    email VARCHAR(255) NOT NULL,
                    code VARCHAR(10) NOT NULL,
                    type VARCHAR(50) DEFAULT 'register',
                    expires_at TIMESTAMP NOT NULL,
                    used BOOLEAN DEFAULT FALSE,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建captcha_sessions表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS captcha_sessions (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    session_id VARCHAR(255) UNIQUE NOT NULL,
                    captcha_text VARCHAR(10) NOT NULL,
                    expires_at TIMESTAMP NOT NULL,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建cards表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS cards (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    name VARCHAR(255) NOT NULL,
                    type VARCHAR(50) NOT NULL,
                    api_config TEXT,
                    text_content TEXT,
                    data_content LONGTEXT,
                    image_url TEXT,
                    description TEXT,
                    enabled BOOLEAN DEFAULT TRUE,
                    delay_seconds INT DEFAULT 0,
                    is_multi_spec BOOLEAN DEFAULT FALSE,
                    spec_name VARCHAR(255),
                    spec_value VARCHAR(255),
                    user_id INT,
                    stock_warning_threshold INT DEFAULT 0 COMMENT '库存预警数量，0表示不启用预警',
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建delivery_rules表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS delivery_rules (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    item_id VARCHAR(255) NOT NULL,
                    card_id INT NOT NULL,
                    delivery_count INT DEFAULT 1,
                    delivery_times INT DEFAULT 0,
                    enabled BOOLEAN DEFAULT TRUE,
                    description TEXT,
                    user_id INT,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                    FOREIGN KEY (card_id) REFERENCES cards(id) ON DELETE CASCADE
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建items表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS items (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    cookie_id VARCHAR(255) NOT NULL,
                    item_id VARCHAR(255) NOT NULL,
                    item_title TEXT,
                    item_description TEXT,
                    item_category VARCHAR(255),
                    item_price VARCHAR(100),
                    item_detail TEXT,
                    is_multi_spec BOOLEAN DEFAULT FALSE,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                    FOREIGN KEY (cookie_id) REFERENCES cookies(id) ON DELETE CASCADE,
                    UNIQUE KEY unique_item (cookie_id, item_id)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建item_info表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS item_info (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    cookie_id VARCHAR(255) NOT NULL,
                    item_id VARCHAR(255) NOT NULL,
                    item_title TEXT,
                    item_description TEXT,
                    item_category VARCHAR(255),
                    item_price VARCHAR(100),
                    item_detail TEXT,
                    is_multi_spec BOOLEAN DEFAULT FALSE,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                    FOREIGN KEY (cookie_id) REFERENCES cookies(id) ON DELETE CASCADE,
                    UNIQUE KEY unique_item_info (cookie_id, item_id)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建user_settings表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS user_settings (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    user_id INT NOT NULL,
                    `key` VARCHAR(255) NOT NULL,
                    value TEXT,
                    description TEXT,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
                    UNIQUE KEY unique_user_setting (user_id, `key`)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建keyword_libraries表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS keyword_libraries (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    user_id INT NOT NULL,
                    name VARCHAR(255) NOT NULL,
                    description TEXT,
                    share_count INT DEFAULT 0,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建keyword_library_items表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS keyword_library_items (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    library_id INT NOT NULL,
                    keyword TEXT NOT NULL,
                    reply TEXT NOT NULL,
                    item_id VARCHAR(255),
                    type VARCHAR(50) DEFAULT 'text',
                    image_url TEXT,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    FOREIGN KEY (library_id) REFERENCES keyword_libraries(id) ON DELETE CASCADE
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建global_keywords表（通用词库）
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS global_keywords (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    user_id INT NOT NULL,
                    keyword TEXT NOT NULL,
                    reply TEXT NOT NULL,
                    type VARCHAR(50) DEFAULT 'text',
                    image_url TEXT,
                    enabled BOOLEAN DEFAULT TRUE,
                    is_fuzzy BOOLEAN DEFAULT FALSE,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
                    UNIQUE KEY unique_global_keyword (user_id, keyword(191))
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建default_reply_records表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS default_reply_records (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    cookie_id VARCHAR(255) NOT NULL,
                    chat_id VARCHAR(255) NOT NULL,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    FOREIGN KEY (cookie_id) REFERENCES cookies(id) ON DELETE CASCADE,
                    UNIQUE KEY unique_default_reply (cookie_id, chat_id)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建unmatch_reply_counts表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS unmatch_reply_counts (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    cookie_id VARCHAR(255) NOT NULL,
                    user_id VARCHAR(255) NOT NULL,
                    chat_id VARCHAR(255) NOT NULL,
                    count INT DEFAULT 1,
                    final_reminder_sent BOOLEAN DEFAULT FALSE,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                    FOREIGN KEY (cookie_id) REFERENCES cookies(id) ON DELETE CASCADE,
                    UNIQUE KEY unique_unmatch_count (cookie_id, user_id, chat_id)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建delayed_guide_tasks表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS delayed_guide_tasks (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    cookie_id VARCHAR(255) NOT NULL,
                    user_id VARCHAR(255) NOT NULL,
                    chat_id VARCHAR(255) NOT NULL,
                    item_id VARCHAR(255),
                    trigger_time TIMESTAMP NULL,
                    scheduled_time TIMESTAMP NOT NULL,
                    status VARCHAR(50) DEFAULT 'pending',
                    sent_at TIMESTAMP NULL,
                    cancelled_at TIMESTAMP NULL,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    FOREIGN KEY (cookie_id) REFERENCES cookies(id) ON DELETE CASCADE
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建pending_payment_checks表 - 存储待付款订单的定时检查任务
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS pending_payment_checks (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    cookie_id VARCHAR(255) NOT NULL,
                    order_id VARCHAR(255) NOT NULL,
                    user_id VARCHAR(255) NOT NULL,
                    item_id VARCHAR(255) NOT NULL,
                    trigger_time TIMESTAMP NOT NULL,
                    scheduled_time TIMESTAMP NOT NULL,
                    status VARCHAR(50) DEFAULT 'pending',
                    checked_at TIMESTAMP NULL,
                    cancelled_at TIMESTAMP NULL,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    FOREIGN KEY (cookie_id) REFERENCES cookies(id) ON DELETE CASCADE,
                    UNIQUE KEY unique_order_check (cookie_id, order_id)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建ai_reply_settings表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS ai_reply_settings (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    cookie_id VARCHAR(255) NOT NULL,
                    enabled BOOLEAN DEFAULT FALSE,
                    api_key TEXT,
                    model VARCHAR(100) DEFAULT 'gpt-3.5-turbo',
                    max_tokens INT DEFAULT 150,
                    temperature DECIMAL(3,2) DEFAULT 0.70,
                    max_bargain_rounds INT DEFAULT 3,
                    max_bargain_percentage DECIMAL(5,2) DEFAULT 10.00,
                    max_bargain_amount DECIMAL(10,2) DEFAULT 50.00,
                    custom_prompts TEXT,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                    FOREIGN KEY (cookie_id) REFERENCES cookies(id) ON DELETE CASCADE,
                    UNIQUE KEY unique_ai_setting (cookie_id)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建default_replies表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS default_replies (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    cookie_id VARCHAR(255) NOT NULL,
                    enabled BOOLEAN DEFAULT FALSE,
                    reply_content TEXT,
                    reply_once BOOLEAN DEFAULT FALSE,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                    FOREIGN KEY (cookie_id) REFERENCES cookies(id) ON DELETE CASCADE,
                    UNIQUE KEY unique_default_reply (cookie_id)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建cookie_status表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS cookie_status (
                    cookie_id VARCHAR(255) PRIMARY KEY,
                    enabled BOOLEAN DEFAULT TRUE,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                    FOREIGN KEY (cookie_id) REFERENCES cookies(id) ON DELETE CASCADE
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建chat_first_replies表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS chat_first_replies (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    cookie_id VARCHAR(255) NOT NULL,
                    chat_id VARCHAR(255) NOT NULL,
                    has_replied BOOLEAN DEFAULT FALSE,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                    FOREIGN KEY (cookie_id) REFERENCES cookies(id) ON DELETE CASCADE,
                    UNIQUE KEY unique_chat_reply (cookie_id, chat_id)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建item_exclusive_replies表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS item_exclusive_replies (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    cookie_id VARCHAR(255) NOT NULL,
                    item_id VARCHAR(255) NOT NULL,
                    reply_content TEXT NOT NULL,
                    enabled BOOLEAN DEFAULT TRUE,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                    FOREIGN KEY (cookie_id) REFERENCES cookies(id) ON DELETE CASCADE,
                    UNIQUE KEY unique_item_reply (cookie_id, item_id)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建ai_conversations表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS ai_conversations (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    cookie_id VARCHAR(255) NOT NULL,
                    chat_id VARCHAR(255) NOT NULL,
                    user_id VARCHAR(255) NOT NULL,
                    item_id VARCHAR(255),
                    role VARCHAR(50) NOT NULL,
                    content TEXT NOT NULL,
                    intent VARCHAR(100),
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    FOREIGN KEY (cookie_id) REFERENCES cookies(id) ON DELETE CASCADE
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建ai_item_cache表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS ai_item_cache (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    cookie_id VARCHAR(255) NOT NULL,
                    item_id VARCHAR(255) NOT NULL,
                    item_info TEXT NOT NULL,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                    FOREIGN KEY (cookie_id) REFERENCES cookies(id) ON DELETE CASCADE,
                    UNIQUE KEY unique_ai_cache (cookie_id, item_id)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建orders表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS orders (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    cookie_id VARCHAR(255) NOT NULL,
                    order_id VARCHAR(255) NOT NULL,
                    buyer_id VARCHAR(255) NOT NULL,
                    item_id VARCHAR(255),
                    item_title TEXT,
                    order_status VARCHAR(50) NOT NULL,
                    order_amount DECIMAL(10,2),
                    buyer_message TEXT,
                    seller_message TEXT,
                    shipping_info TEXT,
                    payment_time TIMESTAMP NULL,
                    shipping_time TIMESTAMP NULL,
                    delivery_time TIMESTAMP NULL,
                    refund_time TIMESTAMP NULL,
                    raw_message TEXT,
                    spec_name VARCHAR(255),
                    spec_value VARCHAR(255),
                    quantity VARCHAR(50),
                    amount VARCHAR(50),
                    coupon_code VARCHAR(100),
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                    FOREIGN KEY (cookie_id) REFERENCES cookies(id) ON DELETE CASCADE,
                    UNIQUE KEY unique_order (cookie_id, order_id)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建blacklist_records表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS blacklist_records (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    cookie_id VARCHAR(255) NOT NULL,
                    session_id VARCHAR(255) NOT NULL,
                    user_id VARCHAR(255),
                    user_name VARCHAR(255),
                    reason VARCHAR(255),
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    FOREIGN KEY (cookie_id) REFERENCES cookies(id) ON DELETE CASCADE
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建聚合豆余额表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS user_balance (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    user_id INT NOT NULL UNIQUE,
                    balance DECIMAL(10,2) DEFAULT 0.00 COMMENT '聚合豆余额',
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
                    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                # 创建聚合豆消费记录表
                cursor.execute('''
                CREATE TABLE IF NOT EXISTS balance_records (
                    id INT AUTO_INCREMENT PRIMARY KEY,
                    user_id INT NOT NULL,
                    order_id VARCHAR(255) COMMENT '闲鱼订单号',
                    amount DECIMAL(10,2) NOT NULL COMMENT '变动金额，正数为充值，负数为消费',
                    balance_after DECIMAL(10,2) NOT NULL COMMENT '操作后余额',
                    record_type VARCHAR(50) NOT NULL COMMENT '记录类型：recharge充值, consume消费, admin_adjust管理员调整',
                    description TEXT COMMENT '变动描述',
                    admin_user_id INT COMMENT '操作管理员ID（如果是管理员操作）',
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
                    FOREIGN KEY (admin_user_id) REFERENCES users(id) ON DELETE SET NULL,
                    INDEX idx_user_id (user_id),
                    INDEX idx_order_id (order_id),
                    INDEX idx_created_at (created_at DESC)
                ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
                ''')

                self.conn.commit()
                logger.info("MySQL数据库初始化完成")

                # 创建性能优化索引
                self._create_performance_indexes(cursor)

                # 检查并升级数据库结构
                self.check_and_upgrade_db(cursor)

            except Exception as e:
                logger.error(f"MySQL数据库初始化失败: {e}")
                raise 
    
    def _create_performance_indexes(self, cursor):
        """创建性能优化索引"""
        try:
            logger.info("开始创建性能优化索引...")
            
            # MySQL索引创建（不支持IF NOT EXISTS，需要用异常处理）
            indexes = [
                # 1. cookies表索引
                ('idx_cookies_user_id', 'CREATE INDEX idx_cookies_user_id ON cookies(user_id)'),
                
                # 2. keywords表索引
                ('idx_keywords_cookie_id', 'CREATE INDEX idx_keywords_cookie_id ON keywords(cookie_id)'),
                ('idx_keywords_item_id', 'CREATE INDEX idx_keywords_item_id ON keywords(item_id)'),
                ('idx_keywords_type', 'CREATE INDEX idx_keywords_type ON keywords(type)'),
                ('idx_keywords_lookup', 'CREATE INDEX idx_keywords_lookup ON keywords(cookie_id, keyword(191), item_id)'),
                
                # 3. item_info表索引
                ('idx_item_info_cookie_id', 'CREATE INDEX idx_item_info_cookie_id ON item_info(cookie_id)'),
                ('idx_item_info_created_at', 'CREATE INDEX idx_item_info_created_at ON item_info(created_at DESC)'),
                
                # 4. default_replies表索引
                ('idx_default_replies_cookie_id', 'CREATE INDEX idx_default_replies_cookie_id ON default_replies(cookie_id)'),
                
                # 5. notification相关表索引
                ('idx_message_notifications_cookie_id', 'CREATE INDEX idx_message_notifications_cookie_id ON message_notifications(cookie_id)'),
                ('idx_notification_channels_user_id', 'CREATE INDEX idx_notification_channels_user_id ON notification_channels(user_id)'),
                
                # 6. 用户相关表索引
                ('idx_users_username', 'CREATE INDEX idx_users_username ON users(username)'),
                ('idx_users_email', 'CREATE INDEX idx_users_email ON users(email)'),
                
                # 7. 日志相关表索引
                ('idx_permanently_paused_chats_cookie_id', 'CREATE INDEX idx_permanently_paused_chats_cookie_id ON permanently_paused_chats(cookie_id)'),
                ('idx_default_reply_records_cookie_id', 'CREATE INDEX idx_default_reply_records_cookie_id ON default_reply_records(cookie_id)'),
                
                # 8. cookie_status表索引
                ('idx_cookie_status_enabled', 'CREATE INDEX idx_cookie_status_enabled ON cookie_status(enabled)'),
                
                # 9. orders表索引
                ('idx_orders_cookie_id', 'CREATE INDEX idx_orders_cookie_id ON orders(cookie_id)'),
                ('idx_orders_buyer_id', 'CREATE INDEX idx_orders_buyer_id ON orders(buyer_id)'),
                ('idx_orders_item_id', 'CREATE INDEX idx_orders_item_id ON orders(item_id)'),
                ('idx_orders_status', 'CREATE INDEX idx_orders_status ON orders(order_status)'),
                ('idx_orders_created_at', 'CREATE INDEX idx_orders_created_at ON orders(created_at DESC)'),
                
                # 10. ai相关表索引
                ('idx_ai_conversations_cookie_id', 'CREATE INDEX idx_ai_conversations_cookie_id ON ai_conversations(cookie_id)'),
                ('idx_ai_conversations_chat_id', 'CREATE INDEX idx_ai_conversations_chat_id ON ai_conversations(chat_id)'),
                ('idx_ai_item_cache_cookie_id', 'CREATE INDEX idx_ai_item_cache_cookie_id ON ai_item_cache(cookie_id)'),
                
                # 11. 延迟任务表索引
                ('idx_delayed_guide_tasks_cookie_id', 'CREATE INDEX idx_delayed_guide_tasks_cookie_id ON delayed_guide_tasks(cookie_id)'),
                ('idx_delayed_guide_tasks_status', 'CREATE INDEX idx_delayed_guide_tasks_status ON delayed_guide_tasks(status)'),
                ('idx_delayed_guide_tasks_scheduled_time', 'CREATE INDEX idx_delayed_guide_tasks_scheduled_time ON delayed_guide_tasks(scheduled_time)')
            ]
            
            # 创建索引，如果已存在则忽略错误
            for index_name, create_sql in indexes:
                try:
                    cursor.execute(create_sql)
                except pymysql.Error as e:
                    if "Duplicate key name" in str(e) or "already exists" in str(e):
                        # 索引已存在，忽略
                        continue
                    else:
                        logger.warning(f"创建索引 {index_name} 失败: {e}")
                        continue
            
            self.conn.commit()
            logger.info("性能优化索引创建完成")
            
        except Exception as e:
            logger.error(f"创建性能优化索引失败: {e}")
            # 索引创建失败不应该影响数据库初始化
            pass

    def check_and_upgrade_db(self, cursor):
        """检查并升级数据库结构"""
        try:
            logger.info("开始检查数据库结构...")
            
            # 初始化默认系统设置
            self._init_default_system_settings(cursor)
            
            # 执行各种升级操作
            self.upgrade_orders_table(cursor)
            self.upgrade_orders_table_add_coupon_code(cursor)
            self.update_admin_user_id(cursor)
            self.upgrade_notification_channels_table(cursor)
            self.upgrade_notification_channels_types(cursor)
            self.migrate_legacy_data(cursor)
            self._migrate_keywords_table_constraints(cursor)
            self.upgrade_keywords_table_for_image_support(cursor)
            self.upgrade_unmatch_reply_counts_table(cursor)
            self.upgrade_missing_columns(cursor)
            self.upgrade_cookies_notification_wxid(cursor)
            
            # 初始化角色权限系统
            self._init_role_system(cursor)
            
            # 升级通用词库表以支持用户分离
            self.upgrade_global_keywords_table(cursor)
            
            # 升级keywords表以支持词库来源记录
            self.upgrade_keywords_table_add_library_id(cursor)
            
            logger.info("数据库结构检查完成")
            
        except Exception as e:
            logger.error(f"数据库结构检查失败: {e}")
            raise

    def _init_default_system_settings(self, cursor):
        """初始化默认系统设置"""
        try:
            default_settings = [
                ('app_version', '2.0.0', '应用版本号'),
                ('db_version', '2.0.0', '数据库版本号'),
                ('enable_registration', 'true', '是否启用用户注册'),
                ('smtp_host', '', 'SMTP服务器地址'),
                ('smtp_port', '587', 'SMTP端口'),
                ('smtp_username', '', 'SMTP用户名'),
                ('smtp_password', '', 'SMTP密码'),
                ('smtp_use_tls', 'true', '是否使用TLS加密'),
            ]
            
            for key, value, description in default_settings:
                cursor.execute('''
                    INSERT IGNORE INTO system_settings (`key`, value, description) 
                    VALUES (%s, %s, %s)
                ''', (key, value, description))
            
            self.conn.commit()
            logger.info("默认系统设置初始化完成")
            
        except Exception as e:
            logger.error(f"初始化默认系统设置失败: {e}")
            self.conn.rollback() 

    def upgrade_global_keywords_table(self, cursor):
        """升级通用词库表以支持用户分离"""
        try:
            logger.info("开始升级通用词库表...")
            
            # 检查是否已有user_id字段
            cursor.execute("SHOW COLUMNS FROM global_keywords")
            columns = [row[0] for row in cursor.fetchall()]
            
            if 'user_id' not in columns:
                # 添加user_id字段
                cursor.execute("ALTER TABLE global_keywords ADD COLUMN user_id INT")
                logger.info("global_keywords表user_id字段添加成功")
                
                # 获取admin用户ID作为默认值
                cursor.execute("SELECT id FROM users WHERE username = 'admin'")
                admin_user = cursor.fetchone()
                admin_user_id = admin_user[0] if admin_user else 1
                
                # 为现有记录设置默认user_id
                cursor.execute("UPDATE global_keywords SET user_id = %s WHERE user_id IS NULL", (admin_user_id,))
                logger.info(f"为现有通用词库记录设置默认用户ID: {admin_user_id}")
                
                # 设置NOT NULL约束
                cursor.execute("ALTER TABLE global_keywords MODIFY user_id INT NOT NULL")
                
                # 添加外键约束
                try:
                    cursor.execute("ALTER TABLE global_keywords ADD FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE")
                    logger.info("global_keywords表外键约束添加成功")
                except Exception as e:
                    logger.warning(f"添加外键约束失败（可能已存在）: {e}")
                
                # 删除旧的唯一约束
                try:
                    cursor.execute("ALTER TABLE global_keywords DROP INDEX unique_global_keyword")
                    logger.info("删除旧的唯一约束成功")
                except Exception as e:
                    logger.warning(f"删除旧唯一约束失败（可能不存在）: {e}")
                
                # 添加新的唯一约束
                try:
                    cursor.execute("ALTER TABLE global_keywords ADD UNIQUE KEY unique_global_keyword (user_id, keyword(191))")
                    logger.info("添加新的唯一约束成功")
                except Exception as e:
                    logger.warning(f"添加新唯一约束失败（可能已存在）: {e}")
            
            # 检查是否已有is_fuzzy字段
            if 'is_fuzzy' not in columns:
                cursor.execute("ALTER TABLE global_keywords ADD COLUMN is_fuzzy BOOLEAN DEFAULT FALSE")
                logger.info("global_keywords表is_fuzzy字段添加成功")
            
            self.conn.commit()
            logger.info("通用词库表升级完成")
            
        except Exception as e:
            logger.error(f"升级通用词库表失败: {e}")
            self.conn.rollback()

    def upgrade_keywords_table_add_library_id(self, cursor):
        """升级keywords表以支持词库来源记录"""
        try:
            logger.info("开始升级keywords表以支持词库来源记录...")
            
            # 检查是否已有library_id字段
            cursor.execute("SHOW COLUMNS FROM keywords")
            columns = [row[0] for row in cursor.fetchall()]
            
            if 'library_id' not in columns:
                # 添加library_id字段
                cursor.execute("ALTER TABLE keywords ADD COLUMN library_id INT DEFAULT NULL")
                logger.info("keywords表library_id字段添加成功")
                
                # 添加外键约束
                try:
                    cursor.execute("ALTER TABLE keywords ADD FOREIGN KEY (library_id) REFERENCES keyword_libraries(id) ON DELETE SET NULL")
                    logger.info("keywords表library_id外键约束添加成功")
                except Exception as e:
                    logger.warning(f"添加library_id外键约束失败（可能已存在）: {e}")
            
            self.conn.commit()
            logger.info("keywords表升级完成")
            
        except Exception as e:
            logger.error(f"升级keywords表失败: {e}")
            self.conn.rollback()

    def _init_role_system(self, cursor):
        """初始化角色权限系统"""
        try:
            logger.info("开始初始化角色权限系统...")
            
            # 初始化默认权限
            default_permissions = [
                ('dashboard_view', '查看仪表盘', 'general'),
                ('accounts_manage', '管理账号', 'account'),
                ('items_manage', '管理商品', 'item'),
                ('orders_manage', '管理订单', 'order'),
                ('auto_reply_manage', '管理自动回复', 'reply'),
                ('auto_reply_item_view', '查看已配置关键词的商品', 'reply'),
                ('keyword_library_manage', '管理门店词库', 'keyword'),
                ('global_keywords_manage', '管理通用词库', 'keyword'),
                ('cards_manage', '管理卡种', 'card'),
                ('auto_delivery_manage', '管理自动发货', 'delivery'),
                ('notification_channels_manage', '管理通知渠道', 'notification'),
                ('message_notifications_manage', '管理消息通知', 'notification'),
                ('system_settings_manage', '管理系统设置', 'system'),
                ('user_management_access', '访问用户管理', 'admin'),
                ('log_management_access', '访问日志管理', 'admin'),
                ('data_management_access', '访问数据管理', 'admin'),
            ]
            
            for name, description, category in default_permissions:
                cursor.execute('''
                    INSERT IGNORE INTO system_permissions (name, description, category) 
                    VALUES (%s, %s, %s)
                ''', (name, description, category))
            
            # 初始化默认角色
            default_roles = [
                ('admin', '系统管理员', True),
                ('user', '普通用户', True),
                ('operator', '运营人员', True),
            ]
            
            for name, description, is_system in default_roles:
                cursor.execute('''
                    INSERT IGNORE INTO roles (name, description, is_system) 
                    VALUES (%s, %s, %s)
                ''', (name, description, is_system))
            
            # 初始化默认菜单
            default_menus = [
                ('dashboard', '仪表盘', 'bi-speedometer2', '#', None, 1, True),
                ('accounts', '账号管理', 'bi-person-circle', '#', None, 2, True),
                ('items', '商品管理', 'bi-box-seam', '#', None, 3, True),
                ('orders', '订单管理', 'bi-receipt', '#', None, 4, True),
                ('auto-reply', '自动回复', 'bi-chat-left-text', '#', None, 5, True),
                ('keyword-library', '门店词库', 'bi-collection', '#', None, 6, True),
                ('global-keywords', '通用词库', 'bi-globe', '#', None, 7, True),
                ('cards', '卡种管理', 'bi-credit-card', '#', None, 8, True),
                ('auto-delivery', '自动发货', 'bi-truck', '#', None, 9, True),
                ('notification-channels', '通知渠道', 'bi-bell', '#', None, 10, True),
                ('message-notifications', '消息通知', 'bi-chat-dots', '#', None, 11, True),
                ('system-settings', '系统设置', 'bi-gear', '#', None, 12, True),
                ('admin-section', '管理员功能', 'bi-shield', '#', None, 100, True),
                ('user-management', '用户管理', 'bi-people', '/user_management.html', None, 101, True),
                ('log-management', '系统日志', 'bi-file-text-fill', '/log_management.html', None, 102, True),
                ('data-management', '数据管理', 'bi-database', '/data_management.html', None, 103, True),
                ('about', '关于', 'bi-info-circle', '#', None, 200, True),
            ]
            
            for menu_key, name, icon, url, parent_id, sort_order, is_system in default_menus:
                cursor.execute('''
                    INSERT IGNORE INTO menus (menu_key, name, icon, url, parent_id, sort_order, is_system) 
                    VALUES (%s, %s, %s, %s, %s, %s, %s)
                ''', (menu_key, name, icon, url, parent_id, sort_order, is_system))
            
            # 获取角色和权限ID
            cursor.execute("SELECT id, name FROM roles")
            roles = {name: id for id, name in cursor.fetchall()}
            
            cursor.execute("SELECT id, name FROM system_permissions") 
            permissions = {name: id for id, name in cursor.fetchall()}
            
            cursor.execute("SELECT id, menu_key FROM menus")
            menus = {menu_key: id for id, menu_key in cursor.fetchall()}
            
            # 为管理员角色分配所有权限
            if 'admin' in roles:
                admin_role_id = roles['admin']
                for perm_id in permissions.values():
                    cursor.execute('''
                        INSERT IGNORE INTO role_permissions (role_id, permission_id) 
                        VALUES (%s, %s)
                    ''', (admin_role_id, perm_id))
                
                # 为管理员角色分配所有菜单
                for menu_id in menus.values():
                    cursor.execute('''
                        INSERT IGNORE INTO role_menus (role_id, menu_id) 
                        VALUES (%s, %s)
                    ''', (admin_role_id, menu_id))
            
            # 为普通用户角色分配基础权限
            if 'user' in roles:
                user_role_id = roles['user']
                user_permissions = [
                    'dashboard_view', 'accounts_manage', 'items_manage', 'orders_manage',
                    'auto_reply_item_view', 'keyword_library_manage', 'cards_manage',
                    'auto_delivery_manage', 'notification_channels_manage', 
                    'message_notifications_manage'
                ]
                for perm_name in user_permissions:
                    if perm_name in permissions:
                        cursor.execute('''
                            INSERT IGNORE INTO role_permissions (role_id, permission_id) 
                            VALUES (%s, %s)
                        ''', (user_role_id, permissions[perm_name]))
                
                # 为普通用户分配基础菜单
                user_menus = [
                    'dashboard', 'accounts', 'items', 'orders', 'auto-reply',
                    'keyword-library', 'cards', 'auto-delivery', 'notification-channels',
                    'message-notifications', 'about'
                ]
                for menu_key in user_menus:
                    if menu_key in menus:
                        cursor.execute('''
                            INSERT IGNORE INTO role_menus (role_id, menu_id) 
                            VALUES (%s, %s)
                        ''', (user_role_id, menus[menu_key]))
            
            # 为运营人员角色分配权限
            if 'operator' in roles:
                operator_role_id = roles['operator']
                operator_permissions = [
                    'dashboard_view', 'accounts_manage', 'items_manage', 'orders_manage',
                    'auto_reply_manage', 'keyword_library_manage', 'global_keywords_manage',
                    'cards_manage', 'auto_delivery_manage', 'notification_channels_manage', 
                    'message_notifications_manage', 'system_settings_manage'
                ]
                for perm_name in operator_permissions:
                    if perm_name in permissions:
                        cursor.execute('''
                            INSERT IGNORE INTO role_permissions (role_id, permission_id) 
                            VALUES (%s, %s)
                        ''', (operator_role_id, permissions[perm_name]))
                
                # 为运营人员分配菜单
                operator_menus = [
                    'dashboard', 'accounts', 'items', 'orders', 'auto-reply',
                    'keyword-library', 'global-keywords', 'cards', 'auto-delivery',
                    'notification-channels', 'message-notifications',
                    'system-settings', 'about'
                ]
                for menu_key in operator_menus:
                    if menu_key in menus:
                        cursor.execute('''
                            INSERT IGNORE INTO role_menus (role_id, menu_id) 
                            VALUES (%s, %s)
                        ''', (operator_role_id, menus[menu_key]))
            
            # 为现有用户分配默认角色
            cursor.execute("SELECT id, username FROM users")
            users = cursor.fetchall()
            
            for user_id, username in users:
                # 检查用户是否已有角色
                cursor.execute("SELECT COUNT(*) FROM user_roles WHERE user_id = %s", (user_id,))
                has_roles = cursor.fetchone()[0] > 0
                
                if not has_roles:
                    if username == 'admin':
                        # admin用户分配管理员角色
                        if 'admin' in roles:
                            cursor.execute('''
                                INSERT IGNORE INTO user_roles (user_id, role_id) 
                                VALUES (%s, %s)
                            ''', (user_id, roles['admin']))
                    else:
                        # 其他用户分配普通用户角色
                        if 'user' in roles:
                            cursor.execute('''
                                INSERT IGNORE INTO user_roles (user_id, role_id) 
                                VALUES (%s, %s)
                            ''', (user_id, roles['user']))
            
            self.conn.commit()
            logger.info("角色权限系统初始化完成")
            
        except Exception as e:
            logger.error(f"初始化角色权限系统失败: {e}")
            self.conn.rollback()

    def close(self):
        """关闭数据库连接"""
        if self.conn:
            self.conn.close()
            self.conn = None
    
    def get_connection(self):
        """获取数据库连接，如果已关闭则重新连接"""
        if self.conn is None:
            self.conn = self._create_connection()
        else:
            # 检查连接是否还活着
            try:
                self.conn.ping(reconnect=False)
            except:
                # 连接已断开，重新创建
                logger.warning("数据库连接已断开，正在重新连接...")
                try:
                    self.conn.close()
                except:
                    pass
                self.conn = self._create_connection()
        return self.conn

    def _log_sql(self, sql: str, params: tuple = None, operation: str = "EXECUTE"):
        """记录SQL执行日志"""
        if not self.sql_log_enabled:
            return

        # 格式化参数
        params_str = ""
        if params:
            if isinstance(params, (list, tuple)):
                if len(params) > 0:
                    # 限制参数长度，避免日志过长
                    formatted_params = []
                    for param in params:
                        if isinstance(param, str) and len(param) > 100:
                            formatted_params.append(f"{param[:100]}...")
                        else:
                            formatted_params.append(repr(param))
                    params_str = f" | 参数: [{', '.join(formatted_params)}]"

        # 格式化SQL（移除多余空白）
        formatted_sql = ' '.join(sql.split())

        # 根据配置的日志级别输出
        log_message = f"🗄️ SQL {operation}: {formatted_sql}{params_str}"

        if self.sql_log_level == 'DEBUG':
            logger.debug(log_message)
        elif self.sql_log_level == 'INFO':
            logger.info(log_message)
        elif self.sql_log_level == 'WARNING':
            logger.warning(log_message)
        else:
            logger.debug(log_message)

    def _execute_sql(self, cursor, sql: str, params: tuple = None):
        """执行SQL并记录日志"""
        self._log_sql(sql, params, "EXECUTE")
        if params:
            return cursor.execute(sql, params)
        else:
            return cursor.execute(sql)

    def _executemany_sql(self, cursor, sql: str, params_list):
        """批量执行SQL并记录日志"""
        self._log_sql(sql, f"批量执行 {len(params_list)} 条记录", "EXECUTEMANY")
        return cursor.executemany(sql, params_list)
    
    # -------------------- Cookie操作 --------------------
    def save_cookie(self, cookie_id: str, cookie_value: str, user_id: int = None) -> bool:
        """保存Cookie到数据库，如存在则更新"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()

                # 检查Cookie是否已存在
                self._execute_sql(cursor, "SELECT user_id, auto_confirm, remark, pause_duration, pause_enabled, token_status FROM cookies WHERE id = %s", (cookie_id,))
                existing = cursor.fetchone()
                
                if existing:
                    # Cookie已存在，只更新value，保留其他字段，并重置Token状态
                    if user_id is None:
                        user_id = existing[0]  # 保留原有的user_id
                    
                    self._execute_sql(cursor,
                        "UPDATE cookies SET value = %s, user_id = %s, token_status = 'valid', token_failed_at = NULL, updated_at = CURRENT_TIMESTAMP WHERE id = %s",
                        (cookie_value, user_id, cookie_id)
                    )
                    logger.info(f"Cookie更新成功: {cookie_id} (用户ID: {user_id}，保留其他设置，重置Token状态)")
                else:
                    # 新Cookie，使用默认值
                    if user_id is None:
                        # 获取admin用户ID作为默认值
                        self._execute_sql(cursor, "SELECT id FROM users WHERE username = 'admin'")
                        admin_user = cursor.fetchone()
                        user_id = admin_user[0] if admin_user else 1

                    self._execute_sql(cursor,
                        "INSERT INTO cookies (id, value, user_id, auto_confirm, remark, pause_duration, pause_enabled, token_status) VALUES (%s, %s, %s, 1, '', 10, 1, 'valid')",
                        (cookie_id, cookie_value, user_id)
                    )
                    logger.info(f"Cookie新增成功: {cookie_id} (用户ID: {user_id})")

                self.conn.commit()

                # 验证保存结果
                self._execute_sql(cursor, "SELECT user_id FROM cookies WHERE id = %s", (cookie_id,))
                saved_user_id = cursor.fetchone()
                if saved_user_id:
                    logger.info(f"Cookie保存验证: {cookie_id} 实际绑定到用户ID: {saved_user_id[0]}")
                else:
                    logger.error(f"Cookie保存验证失败: {cookie_id} 未找到记录")
                return True
            except Exception as e:
                logger.error(f"Cookie保存失败: {e}")
                self.conn.rollback()
                return False
    
    def delete_cookie(self, cookie_id: str) -> bool:
        """从数据库删除Cookie及其关键字"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                # 删除关联的关键字
                self._execute_sql(cursor, "DELETE FROM keywords WHERE cookie_id = %s", (cookie_id,))
                # 删除Cookie
                self._execute_sql(cursor, "DELETE FROM cookies WHERE id = %s", (cookie_id,))
                self.conn.commit()
                logger.info(f"Cookie删除成功: {cookie_id}")
                return True
            except Exception as e:
                logger.error(f"Cookie删除失败: {e}")
                self.conn.rollback()
                return False

    def get_cookie(self, cookie_id: str) -> Optional[str]:
        """从数据库获取Cookie值"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                self._execute_sql(cursor, "SELECT value FROM cookies WHERE id = %s", (cookie_id,))
                result = cursor.fetchone()
                return result[0] if result else None
            except Exception as e:
                logger.error(f"获取Cookie失败: {e}")
                return None

    def get_all_cookies(self, user_id: int = None) -> Dict[str, str]:
        """获取所有Cookie"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                if user_id is not None:
                    self._execute_sql(cursor, "SELECT id, value FROM cookies WHERE user_id = %s", (user_id,))
                else:
                    self._execute_sql(cursor, "SELECT id, value FROM cookies")
                results = cursor.fetchall()
                return {row[0]: row[1] for row in results}
            except Exception as e:
                logger.error(f"获取所有Cookie失败: {e}")
                return {}

    def get_cookie_by_id(self, cookie_id: str) -> Optional[Dict[str, str]]:
        """根据ID获取Cookie详细信息"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                self._execute_sql(cursor, """
                    SELECT id, value, user_id, remark, auto_confirm, pause_duration, 
                           pause_enabled, token_status, token_failed_at, notification_wxid,
                           created_at, updated_at 
                    FROM cookies WHERE id = %s
                """, (cookie_id,))
                result = cursor.fetchone()
                
                if result:
                    return {
                        'id': result[0],
                        'value': result[1],
                        'user_id': result[2],
                        'remark': result[3],
                        'auto_confirm': bool(result[4]),
                        'pause_duration': result[5],
                        'pause_enabled': bool(result[6]),
                        'token_status': result[7],
                        'token_failed_at': result[8],
                        'notification_wxid': result[9],
                        'created_at': result[10],
                        'updated_at': result[11]
                    }
                return None
            except Exception as e:
                logger.error(f"根据ID获取Cookie失败: {e}")
                return None 

    # -------------------- 关键字操作 --------------------
    def save_keywords(self, cookie_id: str, keywords: List[Tuple[str, str]]) -> bool:
        """保存关键字列表，先删除旧数据再插入新数据（向后兼容方法）"""
        # 转换为新格式（不包含item_id）
        keywords_with_item_id = [(keyword, reply, None) for keyword, reply in keywords]
        return self.save_keywords_with_item_id(cookie_id, keywords_with_item_id)

    def save_keywords_with_item_id(self, cookie_id: str, keywords: List[Tuple[str, str, str]]) -> bool:
        """保存关键字列表（包含商品ID），先删除旧数据再插入新数据"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()

                # 先删除该cookie_id的所有关键字
                self._execute_sql(cursor, "DELETE FROM keywords WHERE cookie_id = %s", (cookie_id,))

                # 插入新关键字，使用INSERT IGNORE来处理可能的唯一约束冲突
                for keyword, reply, item_id in keywords:
                    # 标准化item_id：空字符串转为NULL
                    normalized_item_id = item_id if item_id and item_id.strip() else None

                    try:
                        self._execute_sql(cursor,
                            "INSERT INTO keywords (cookie_id, keyword, reply, item_id) VALUES (%s, %s, %s, %s)",
                            (cookie_id, keyword, reply, normalized_item_id))
                    except pymysql.IntegrityError as ie:
                        # 如果遇到唯一约束冲突，记录详细错误信息
                        item_desc = f"商品ID: {normalized_item_id}" if normalized_item_id else "通用关键词"
                        logger.error(f"关键词唯一约束冲突: Cookie={cookie_id}, 关键词='{keyword}', {item_desc}")
                        raise ie

                self.conn.commit()
                logger.info(f"关键字保存成功: {cookie_id}, {len(keywords)}条")
                return True
            except Exception as e:
                logger.error(f"关键字保存失败: {e}")
                self.conn.rollback()
                return False

    def save_text_keywords_only(self, cookie_id: str, keywords: List[Tuple[str, str, str]]) -> bool:
        """保存文本关键字列表，只删除文本类型的关键词，保留图片关键词"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()

                # 只删除该cookie_id的文本类型关键字，保留图片关键词
                self._execute_sql(cursor,
                    "DELETE FROM keywords WHERE cookie_id = %s AND (type IS NULL OR type = 'text')",
                    (cookie_id,))

                # 插入新的文本关键字
                for keyword, reply, item_id in keywords:
                    # 标准化item_id：空字符串转为NULL
                    normalized_item_id = item_id if item_id and item_id.strip() else None

                    try:
                        self._execute_sql(cursor,
                            "INSERT INTO keywords (cookie_id, keyword, reply, item_id, type) VALUES (%s, %s, %s, %s, 'text')",
                            (cookie_id, keyword, reply, normalized_item_id))
                    except pymysql.IntegrityError as ie:
                        # 如果遇到唯一约束冲突，记录详细错误信息并回滚
                        item_desc = f"商品ID: {normalized_item_id}" if normalized_item_id else "通用关键词"
                        logger.error(f"关键词唯一约束冲突: Cookie={cookie_id}, 关键词='{keyword}', {item_desc}")
                        self.conn.rollback()
                        raise ie

                self.conn.commit()
                logger.info(f"文本关键字保存成功: {cookie_id}, {len(keywords)}条，图片关键词已保留")
                return True
            except pymysql.IntegrityError:
                # 唯一约束冲突，重新抛出异常让上层处理
                raise
            except Exception as e:
                logger.error(f"文本关键字保存失败: {e}")
                self.conn.rollback()
                return False
    
    def get_keywords(self, cookie_id: str) -> List[Tuple[str, str]]:
        """获取指定Cookie的关键字列表（向后兼容方法）"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                self._execute_sql(cursor, "SELECT keyword, reply FROM keywords WHERE cookie_id = %s", (cookie_id,))
                return [(row[0], row[1]) for row in cursor.fetchall()]
            except Exception as e:
                logger.error(f"获取关键字失败: {e}")
                return []

    def get_keywords_with_item_id(self, cookie_id: str) -> List[Tuple[str, str, str]]:
        """获取指定Cookie的关键字列表（包含商品ID）"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                self._execute_sql(cursor, "SELECT keyword, reply, item_id FROM keywords WHERE cookie_id = %s", (cookie_id,))
                return [(row[0], row[1], row[2]) for row in cursor.fetchall()]
            except Exception as e:
                logger.error(f"获取关键字失败: {e}")
                return []

    def check_keyword_duplicate(self, cookie_id: str, keyword: str, item_id: str = None) -> bool:
        """检查关键词是否重复"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                if item_id:
                    # 如果有商品ID，检查相同cookie_id、keyword、item_id的组合
                    self._execute_sql(cursor,
                        "SELECT COUNT(*) FROM keywords WHERE cookie_id = %s AND keyword = %s AND item_id = %s",
                        (cookie_id, keyword, item_id))
                else:
                    # 如果没有商品ID，检查相同cookie_id、keyword且item_id为空的组合
                    self._execute_sql(cursor,
                        "SELECT COUNT(*) FROM keywords WHERE cookie_id = %s AND keyword = %s AND (item_id IS NULL OR item_id = '')",
                        (cookie_id, keyword))

                count = cursor.fetchone()[0]
                return count > 0
            except Exception as e:
                logger.error(f"检查关键词重复失败: {e}")
                return False

    def save_image_keyword(self, cookie_id: str, keyword: str, image_url: str, item_id: str = None) -> bool:
        """保存图片关键词（调用前应先检查重复）"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()

                # 标准化item_id：空字符串转为NULL
                normalized_item_id = item_id if item_id and item_id.strip() else None

                # 直接插入图片关键词（重复检查应在调用前完成）
                self._execute_sql(cursor,
                    "INSERT INTO keywords (cookie_id, keyword, reply, item_id, type, image_url) VALUES (%s, %s, %s, %s, %s, %s)",
                    (cookie_id, keyword, '', normalized_item_id, 'image', image_url))

                self.conn.commit()
                logger.info(f"图片关键词保存成功: {cookie_id}, 关键词: {keyword}, 图片: {image_url}")
                return True
            except Exception as e:
                logger.error(f"图片关键词保存失败: {e}")
                self.conn.rollback()
                return False

    def add_keyword(self, cookie_id: str, keyword: str, reply: str, item_id: str = None) -> bool:
        """添加单个文本关键词（调用前应先检查重复）"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()

                # 标准化item_id：空字符串转为NULL
                normalized_item_id = item_id if item_id and item_id.strip() else None

                # 直接插入文本关键词（重复检查应在调用前完成）
                self._execute_sql(cursor,
                    "INSERT INTO keywords (cookie_id, keyword, reply, item_id, type) VALUES (%s, %s, %s, %s, 'text')",
                    (cookie_id, keyword, reply, normalized_item_id))

                self.conn.commit()
                logger.info(f"文本关键词添加成功: {cookie_id}, 关键词: {keyword}, 商品ID: {normalized_item_id}")
                return True
            except Exception as e:
                logger.error(f"文本关键词添加失败: {e}")
                self.conn.rollback()
                return False

    def add_keywords_append(self, cookie_id: str, keywords: List[Tuple[str, str, str]]) -> bool:
        """追加关键词到账号（不删除现有关键词）"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                added_count = 0
                skipped_count = 0

                for keyword, reply, item_id in keywords:
                    # 标准化item_id：空字符串转为NULL
                    normalized_item_id = item_id if item_id and item_id.strip() else None

                    try:
                        self._execute_sql(cursor,
                            "INSERT INTO keywords (cookie_id, keyword, reply, item_id, type) VALUES (%s, %s, %s, %s, 'text')",
                            (cookie_id, keyword, reply, normalized_item_id))
                        added_count += 1
                    except pymysql.IntegrityError:
                        # 如果遇到唯一约束冲突，跳过这个关键词
                        skipped_count += 1
                        continue
                    except Exception as e:
                        logger.error(f"添加关键词失败: {cookie_id}, {keyword}, {e}")
                        continue

                self.conn.commit()
                logger.info(f"关键词追加成功: {cookie_id}, 新增: {added_count}, 跳过重复: {skipped_count}")
                return True
            except Exception as e:
                logger.error(f"关键词追加失败: {e}")
                self.conn.rollback()
                return False

    def add_keywords_append_with_library(self, cookie_id: str, keywords: List[Tuple[str, str, str]], library_id: int = None) -> bool:
        """追加关键词到账号（不删除现有关键词），支持记录词库来源"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 准备批量插入数据，过滤重复的关键词
                batch_data = []
                duplicate_check = set()
                
                # 首先获取现有关键词，避免重复插入
                existing_query = """
                    SELECT CONCAT(keyword, '|', COALESCE(item_id, '')) as key_combo 
                    FROM keywords 
                    WHERE cookie_id = %s
                """
                cursor.execute(existing_query, (cookie_id,))
                existing_keys = {row[0] for row in cursor.fetchall()}
                
                for keyword, reply, item_id in keywords:
                    # 标准化item_id：空字符串转为NULL
                    normalized_item_id = item_id if item_id and item_id.strip() else None
                    
                    # 创建唯一键用于重复检查
                    key_combo = f"{keyword}|{normalized_item_id or ''}"
                    
                    # 跳过重复的关键词
                    if key_combo not in existing_keys and key_combo not in duplicate_check:
                        batch_data.append((
                            cookie_id, 
                            keyword, 
                            reply, 
                            normalized_item_id, 
                            'text', 
                            library_id
                        ))
                        duplicate_check.add(key_combo)
                
                if not batch_data:
                    logger.info(f"关键词追加跳过: {cookie_id}, 所有关键词都已存在, 词库ID: {library_id}")
                    return True
                
                # 使用executemany进行批量插入
                start_time = time.time()
                cursor.executemany(
                    "INSERT INTO keywords (cookie_id, keyword, reply, item_id, type, library_id) VALUES (%s, %s, %s, %s, %s, %s)",
                    batch_data
                )
                insert_time = time.time() - start_time
                
                self.conn.commit()
                added_count = len(batch_data)
                skipped_count = len(keywords) - added_count
                
                logger.info(f"关键词批量追加成功: {cookie_id}, 新增: {added_count}, 跳过重复: {skipped_count}, 词库ID: {library_id}, 耗时: {insert_time:.2f}s")
                return True
                
            except Exception as e:
                logger.error(f"关键词追加失败: {e}")
                self.conn.rollback()
                return False

    def get_keywords_with_type(self, cookie_id: str) -> List[Dict[str, any]]:
        """获取指定Cookie的关键字列表（包含类型信息和词库信息）"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                self._execute_sql(cursor,
                    """SELECT k.keyword, k.reply, k.item_id, k.type, k.image_url, k.library_id, kl.name as library_name 
                       FROM keywords k 
                       LEFT JOIN keyword_libraries kl ON k.library_id = kl.id 
                       WHERE k.cookie_id = %s""",
                    (cookie_id,))

                results = []
                for row in cursor.fetchall():
                    keyword_data = {
                        'keyword': row[0],
                        'reply': row[1],
                        'item_id': row[2],
                        'type': row[3] or 'text',  # 如果type为NULL，默认为text
                        'image_url': row[4],
                        'library_id': row[5],
                        'library_name': row[6]
                    }
                    results.append(keyword_data)

                return results
            except Exception as e:
                logger.error(f"获取关键字（含类型）失败: {e}")
                return [] 

    def upgrade_orders_table(self, cursor):
        """升级数据库添加订单管理表"""
        try:
            logger.info("开始检查订单管理表...")
            
            # 检查orders表是否已存在
            cursor.execute("SHOW TABLES LIKE 'orders'")
            if cursor.fetchone():
                logger.info("orders表已存在，检查表结构...")
                
                # 检查表结构是否完整
                cursor.execute("SHOW COLUMNS FROM orders")
                columns = [row[0] for row in cursor.fetchall()]
                
                required_columns = [
                    'id', 'cookie_id', 'order_id', 'buyer_id', 'item_id', 'item_title',
                    'order_status', 'order_amount', 'buyer_message', 'seller_message',
                    'shipping_info', 'payment_time', 'shipping_time', 'delivery_time',
                    'refund_time', 'raw_message', 'created_at', 'updated_at'
                ]
                
                missing_columns = [col for col in required_columns if col not in columns]
                
                if missing_columns:
                    logger.info(f"orders表缺少字段: {missing_columns}，添加缺失字段...")
                    for col in missing_columns:
                        if col in ['payment_time', 'shipping_time', 'delivery_time', 'refund_time']:
                            cursor.execute(f"ALTER TABLE orders ADD COLUMN {col} TIMESTAMP NULL")
                        elif col == 'order_amount':
                            cursor.execute(f"ALTER TABLE orders ADD COLUMN {col} DECIMAL(10,2)")
                        elif col in ['created_at', 'updated_at']:
                            if col == 'updated_at':
                                cursor.execute(f"ALTER TABLE orders ADD COLUMN {col} TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP")
                            else:
                                cursor.execute(f"ALTER TABLE orders ADD COLUMN {col} TIMESTAMP DEFAULT CURRENT_TIMESTAMP")
                        else:
                            cursor.execute(f"ALTER TABLE orders ADD COLUMN {col} TEXT")
                    logger.info("orders表字段添加完成")
                else:
                    logger.info("orders表结构完整")
            else:
                logger.info("orders表不存在，已在init_db中创建")
                
            self.conn.commit()
            logger.info("订单表升级完成")
            
        except Exception as e:
            logger.error(f"升级订单表失败: {e}")
            self.conn.rollback()

    def upgrade_orders_table_add_coupon_code(self, cursor):
        """为订单表添加券码字段"""
        try:
            logger.info("开始为订单表添加券码字段...")
            
            # 检查是否已有coupon_code字段
            cursor.execute("SHOW COLUMNS FROM orders")
            columns = [row[0] for row in cursor.fetchall()]
            
            if 'coupon_code' not in columns:
                cursor.execute("ALTER TABLE orders ADD COLUMN coupon_code VARCHAR(100)")
                logger.info("订单表coupon_code字段添加成功")
            else:
                logger.info("订单表coupon_code字段已存在")
                
            self.conn.commit()
            
        except Exception as e:
            logger.error(f"为订单表添加券码字段失败: {e}")
            self.conn.rollback()

    def update_admin_user_id(self, cursor):
        """更新管理员用户ID"""
        try:
            # 检查是否存在admin用户
            cursor.execute("SELECT id FROM users WHERE username = 'admin'")
            admin_user = cursor.fetchone()
            
            if not admin_user:
                # 创建默认admin用户
                password_hash = hashlib.sha256('admin123'.encode()).hexdigest()
                cursor.execute("""
                    INSERT INTO users (username, email, password_hash, is_admin) 
                    VALUES ('admin', 'admin@localhost', %s, TRUE)
                """, (password_hash,))
                logger.info("默认admin用户创建成功")
            else:
                logger.info("admin用户已存在")
                
            self.conn.commit()
            
        except Exception as e:
            logger.error(f"更新管理员用户ID失败: {e}")
            self.conn.rollback()

    def upgrade_notification_channels_table(self, cursor):
        """升级通知渠道表"""
        try:
            # 检查表是否存在
            cursor.execute("SHOW TABLES LIKE 'notification_channels'")
            if not cursor.fetchone():
                logger.info("notification_channels表不存在，已在init_db中创建")
                return
                
            # 检查字段是否完整
            cursor.execute("SHOW COLUMNS FROM notification_channels")
            columns = [row[0] for row in cursor.fetchall()]
            
            if 'user_id' not in columns:
                cursor.execute("ALTER TABLE notification_channels ADD COLUMN user_id INT")
                logger.info("notification_channels表user_id字段添加成功")
                
            self.conn.commit()
            
        except Exception as e:
            logger.error(f"升级通知渠道表失败: {e}")
            self.conn.rollback()

    def upgrade_notification_channels_types(self, cursor):
        """升级通知渠道类型支持"""
        try:
            logger.info("检查通知渠道类型支持...")
            # MySQL中不需要CHECK约束的特殊处理，类型验证在应用层处理
            logger.info("通知渠道类型支持检查完成")
            
        except Exception as e:
            logger.error(f"升级通知渠道类型失败: {e}")

    def migrate_legacy_data(self, cursor):
        """迁移遗留数据"""
        try:
            logger.info("开始迁移遗留数据...")
            
            # 初始化默认系统设置（如果不存在）
            self._init_default_system_settings(cursor)
            
            # 迁移表数据
            self._migrate_table_data(cursor, 'notification_channels')
            
            logger.info("遗留数据迁移完成")
            
        except Exception as e:
            logger.error(f"迁移遗留数据失败: {e}")

    def _migrate_table_data(self, cursor, table_name: str):
        """迁移表数据"""
        try:
            # 标准化channel_type字段
            if table_name == 'notification_channels':
                cursor.execute("SELECT id, channel_type FROM notification_channels")
                channels = cursor.fetchall()
                
                for channel_id, old_type in channels:
                    new_type = self._normalize_channel_type(old_type)
                    if new_type != old_type:
                        cursor.execute(
                            "UPDATE notification_channels SET channel_type = %s WHERE id = %s",
                            (new_type, channel_id)
                        )
                        
                self.conn.commit()
                logger.info(f"{table_name}表数据迁移完成")
                
        except Exception as e:
            logger.error(f"迁移{table_name}表数据失败: {e}")
            self.conn.rollback()

    def _normalize_channel_type(self, old_type: str) -> str:
        """标准化渠道类型"""
        type_mapping = {
            'wechat': 'wechat',
            'email': 'email',
            'webhook': 'webhook',
            'dingtalk': 'dingtalk',
            'feishu': 'feishu'
        }
        return type_mapping.get(old_type.lower(), old_type)

    def _migrate_keywords_table_constraints(self, cursor):
        """迁移关键词表约束"""
        try:
            logger.info("检查关键词表约束...")
            # MySQL中的约束处理
            cursor.execute("SHOW COLUMNS FROM keywords")
            columns = [row[0] for row in cursor.fetchall()]
            
            if 'type' not in columns:
                cursor.execute("ALTER TABLE keywords ADD COLUMN type VARCHAR(50) DEFAULT 'text'")
                logger.info("关键词表type字段添加成功")
                
            if 'image_url' not in columns:
                cursor.execute("ALTER TABLE keywords ADD COLUMN image_url TEXT")
                logger.info("关键词表image_url字段添加成功")
                
            self.conn.commit()
            
        except Exception as e:
            logger.error(f"迁移关键词表约束失败: {e}")
            self.conn.rollback()

    def upgrade_keywords_table_for_image_support(self, cursor):
        """升级关键词表以支持图片"""
        try:
            # 检查字段是否存在
            cursor.execute("SHOW COLUMNS FROM keywords")
            columns = [row[0] for row in cursor.fetchall()]
            
            if 'type' not in columns:
                cursor.execute("ALTER TABLE keywords ADD COLUMN type VARCHAR(50) DEFAULT 'text'")
                logger.info("关键词表type字段添加成功")
                
            if 'image_url' not in columns:
                cursor.execute("ALTER TABLE keywords ADD COLUMN image_url TEXT")
                logger.info("关键词表image_url字段添加成功")
                
            self.conn.commit()
            
        except Exception as e:
            logger.error(f"升级关键词表图片支持失败: {e}")
            self.conn.rollback()

    def upgrade_unmatch_reply_counts_table(self, cursor):
        """升级未匹配回复计数表"""
        try:
            # 检查表是否存在
            cursor.execute("SHOW TABLES LIKE 'unmatch_reply_counts'")
            if not cursor.fetchone():
                logger.info("unmatch_reply_counts表不存在，已在init_db中创建")
                return
                
            logger.info("unmatch_reply_counts表升级完成")
            
        except Exception as e:
            logger.error(f"升级未匹配回复计数表失败: {e}")

    def upgrade_missing_columns(self, cursor):
        """添加缺失的列"""
        try:
            logger.info("检查并添加缺失的列...")
            
            # 检查各表的缺失列
            tables_to_check = {
                'cookies': ['notification_wxid'],
                'cards': ['user_id', 'stock_warning_threshold'],
                'delivery_rules': ['user_id', 'generate_image', 'is_multi_spec', 'spec_name', 'spec_value'],
                'ai_reply_settings': ['custom_prompts'],
            }
            
            for table_name, required_columns in tables_to_check.items():
                try:
                    cursor.execute(f"SHOW COLUMNS FROM {table_name}")
                    existing_columns = [row[0] for row in cursor.fetchall()]
                    
                    for col in required_columns:
                        if col not in existing_columns:
                            if col == 'user_id':
                                cursor.execute(f"ALTER TABLE {table_name} ADD COLUMN {col} INT")
                            elif col == 'generate_image':
                                cursor.execute(f"ALTER TABLE {table_name} ADD COLUMN {col} BOOLEAN DEFAULT TRUE")
                            elif col == 'stock_warning_threshold':
                                cursor.execute(f"ALTER TABLE {table_name} ADD COLUMN {col} INT DEFAULT 0 COMMENT '库存预警数量，0表示不启用预警'")
                            elif col == 'is_multi_spec':
                                cursor.execute(f"ALTER TABLE {table_name} ADD COLUMN {col} BOOLEAN DEFAULT FALSE")
                            elif col in ['spec_name', 'spec_value']:
                                cursor.execute(f"ALTER TABLE {table_name} ADD COLUMN {col} VARCHAR(255)")
                            else:
                                cursor.execute(f"ALTER TABLE {table_name} ADD COLUMN {col} TEXT")
                            logger.info(f"{table_name}表{col}字段添加成功")
                            
                except Exception as e:
                    logger.error(f"检查{table_name}表失败: {e}")
                    continue
                    
            self.conn.commit()
            logger.info("缺失列添加完成")
            
        except Exception as e:
            logger.error(f"添加缺失列失败: {e}")
            self.conn.rollback()

    def upgrade_cookies_notification_wxid(self, cursor):
        """为cookies表添加notification_wxid字段"""
        try:
            cursor.execute("SHOW COLUMNS FROM cookies")
            columns = [row[0] for row in cursor.fetchall()]
            
            if 'notification_wxid' not in columns:
                cursor.execute("ALTER TABLE cookies ADD COLUMN notification_wxid TEXT")
                logger.info("cookies表notification_wxid字段添加成功")
            else:
                logger.info("cookies表notification_wxid字段已存在")
                
            self.conn.commit()
            
        except Exception as e:
            logger.error(f"为cookies表添加notification_wxid字段失败: {e}")
            self.conn.rollback() 

    # -------------------- AI回复设置操作 --------------------
    def save_ai_reply_settings(self, cookie_id: str, settings: dict) -> bool:
        """保存AI回复设置"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 使用INSERT ... ON DUPLICATE KEY UPDATE语法
                cursor.execute('''
                INSERT INTO ai_reply_settings 
                (cookie_id, enabled, api_key, model, max_tokens, temperature, 
                 max_bargain_rounds, max_bargain_percentage, max_bargain_amount, custom_prompts)
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
                ON DUPLICATE KEY UPDATE
                enabled = VALUES(enabled),
                api_key = VALUES(api_key),
                model = VALUES(model),
                max_tokens = VALUES(max_tokens),
                temperature = VALUES(temperature),
                max_bargain_rounds = VALUES(max_bargain_rounds),
                max_bargain_percentage = VALUES(max_bargain_percentage),
                max_bargain_amount = VALUES(max_bargain_amount),
                custom_prompts = VALUES(custom_prompts),
                updated_at = CURRENT_TIMESTAMP
                ''', (
                    cookie_id,
                    settings.get('enabled', False),
                    settings.get('api_key', ''),
                    settings.get('model', 'gpt-3.5-turbo'),
                    settings.get('max_tokens', 150),
                    settings.get('temperature', 0.7),
                    settings.get('max_bargain_rounds', 3),
                    settings.get('max_bargain_percentage', 10.0),
                    settings.get('max_bargain_amount', 50.0),
                    settings.get('custom_prompts', '')
                ))
                
                self.conn.commit()
                logger.info(f"AI回复设置保存成功: {cookie_id}")
                return True
            except Exception as e:
                logger.error(f"保存AI回复设置失败: {e}")
                self.conn.rollback()
                return False

    def get_ai_reply_settings(self, cookie_id: str) -> dict:
        """获取指定账号的AI回复设置"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT enabled, api_key, model, max_tokens, temperature, 
                       max_bargain_rounds, max_bargain_percentage, max_bargain_amount, custom_prompts
                FROM ai_reply_settings WHERE cookie_id = %s
                ''', (cookie_id,))
                
                result = cursor.fetchone()
                if result:
                    return {
                        'enabled': bool(result[0]),
                        'api_key': result[1] or '',
                        'model': result[2] or 'gpt-3.5-turbo',
                        'max_tokens': result[3] or 150,
                        'temperature': float(result[4]) if result[4] else 0.7,
                        'max_bargain_rounds': result[5] or 3,
                        'max_bargain_percentage': float(result[6]) if result[6] else 10.0,
                        'max_bargain_amount': float(result[7]) if result[7] else 50.0,
                        'custom_prompts': result[8] or '',
                        'base_url': 'https://api.openai.com/v1'  # 默认 OpenAI API 地址
                    }
                else:
                    # 返回默认设置
                    return {
                        'enabled': False,
                        'api_key': '',
                        'model': 'gpt-3.5-turbo',
                        'max_tokens': 150,
                        'temperature': 0.7,
                        'max_bargain_rounds': 3,
                        'max_bargain_percentage': 10.0,
                        'max_bargain_amount': 50.0,
                        'custom_prompts': '',
                        'base_url': 'https://api.openai.com/v1'  # 默认 OpenAI API 地址
                    }
            except Exception as e:
                logger.error(f"获取AI回复设置失败: {e}")
                return {
                    'enabled': False,
                    'api_key': '',
                    'model': 'gpt-3.5-turbo',
                    'max_tokens': 150,
                    'temperature': 0.7,
                    'max_bargain_rounds': 3,
                    'max_bargain_percentage': 10.0,
                    'max_bargain_amount': 50.0,
                    'custom_prompts': '',
                    'base_url': 'https://api.openai.com/v1'  # 默认 OpenAI API 地址
                }

    def get_all_ai_reply_settings(self) -> Dict[str, dict]:
        """获取所有账号的AI回复设置"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT cookie_id, enabled, api_key, model, max_tokens, temperature,
                       max_bargain_rounds, max_bargain_percentage, max_bargain_amount, custom_prompts
                FROM ai_reply_settings
                ''')

                result = {}
                for row in cursor.fetchall():
                    cookie_id = row[0]
                    result[cookie_id] = {
                        'enabled': bool(row[1]),
                        'api_key': row[2] or '',
                        'model': row[3] or 'gpt-3.5-turbo',
                        'max_tokens': row[4] or 150,
                        'temperature': float(row[5]) if row[5] else 0.7,
                        'max_bargain_rounds': row[6] or 3,
                        'max_bargain_percentage': float(row[7]) if row[7] else 10.0,
                        'max_bargain_amount': float(row[8]) if row[8] else 50.0,
                        'custom_prompts': row[9] or ''
                    }

                return result
            except Exception as e:
                logger.error(f"获取所有AI回复设置失败: {e}")
                return {}

    # -------------------- 默认回复操作 --------------------
    def save_default_reply(self, cookie_id: str, enabled: bool, reply_content: str = None, reply_once: bool = False):
        """保存默认回复设置"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                INSERT INTO default_replies (cookie_id, enabled, reply_content, reply_once)
                VALUES (%s, %s, %s, %s)
                ON DUPLICATE KEY UPDATE
                enabled = VALUES(enabled),
                reply_content = VALUES(reply_content),
                reply_once = VALUES(reply_once),
                updated_at = CURRENT_TIMESTAMP
                ''', (cookie_id, enabled, reply_content, reply_once))
                self.conn.commit()
                logger.debug(f"保存默认回复设置: {cookie_id} -> {'启用' if enabled else '禁用'}, 只回复一次: {'是' if reply_once else '否'}")
            except Exception as e:
                logger.error(f"保存默认回复设置失败: {e}")
                raise

    def get_default_reply(self, cookie_id: str) -> Optional[Dict[str, any]]:
        """获取指定账号的默认回复设置"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT enabled, reply_content, reply_once FROM default_replies WHERE cookie_id = %s
                ''', (cookie_id,))
                result = cursor.fetchone()
                if result:
                    enabled, reply_content, reply_once = result
                    return {
                        'enabled': bool(enabled),
                        'reply_content': reply_content or '',
                        'reply_once': bool(reply_once) if reply_once is not None else False
                    }
                return None
            except Exception as e:
                logger.error(f"获取默认回复设置失败: {e}")
                return None

    def get_all_default_replies(self) -> Dict[str, Dict[str, any]]:
        """获取所有账号的默认回复设置"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('SELECT cookie_id, enabled, reply_content, reply_once FROM default_replies')

                result = {}
                for row in cursor.fetchall():
                    cookie_id, enabled, reply_content, reply_once = row
                    result[cookie_id] = {
                        'enabled': bool(enabled),
                        'reply_content': reply_content or '',
                        'reply_once': bool(reply_once) if reply_once is not None else False
                    }

                return result
            except Exception as e:
                logger.error(f"获取所有默认回复设置失败: {e}")
                return {}

    def add_default_reply_record(self, cookie_id: str, chat_id: str):
        """记录已回复的chat_id"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                INSERT IGNORE INTO default_reply_records (cookie_id, chat_id)
                VALUES (%s, %s)
                ''', (cookie_id, chat_id))
                self.conn.commit()
                logger.debug(f"记录默认回复: {cookie_id} -> {chat_id}")
            except Exception as e:
                logger.error(f"记录默认回复失败: {e}")

    def has_default_reply_record(self, cookie_id: str, chat_id: str) -> bool:
        """检查是否已经回复过该chat_id"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT 1 FROM default_reply_records WHERE cookie_id = %s AND chat_id = %s
                ''', (cookie_id, chat_id))
                result = cursor.fetchone()
                return result is not None
            except Exception as e:
                logger.error(f"检查默认回复记录失败: {e}")
                return False

    def clear_default_reply_records(self, cookie_id: str):
        """清空指定账号的默认回复记录"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('DELETE FROM default_reply_records WHERE cookie_id = %s', (cookie_id,))
                self.conn.commit()
                logger.debug(f"清空默认回复记录: {cookie_id}")
            except Exception as e:
                logger.error(f"清空默认回复记录失败: {e}")

    def delete_default_reply(self, cookie_id: str) -> bool:
        """删除指定账号的默认回复设置"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                self._execute_sql(cursor, "DELETE FROM default_replies WHERE cookie_id = %s", (cookie_id,))
                self.conn.commit()
                logger.debug(f"删除默认回复设置: {cookie_id}")
                return True
            except Exception as e:
                logger.error(f"删除默认回复设置失败: {e}")
                self.conn.rollback()
                return False

    # -------------------- 通知渠道操作 --------------------
    def create_notification_channel(self, name: str, channel_type: str, config: str, user_id: int = None) -> int:
        """创建通知渠道"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                INSERT INTO notification_channels (name, channel_type, config, user_id)
                VALUES (%s, %s, %s, %s)
                ''', (name, channel_type, config, user_id))
                self.conn.commit()
                channel_id = cursor.lastrowid
                logger.debug(f"创建通知渠道: {name} (ID: {channel_id})")
                return channel_id
            except Exception as e:
                logger.error(f"创建通知渠道失败: {e}")
                self.conn.rollback()
                raise

    def get_notification_channels(self, user_id: int = None) -> List[Dict[str, any]]:
        """获取所有通知渠道"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                if user_id is not None:
                    cursor.execute('''
                    SELECT id, name, channel_type, config, enabled, created_at, updated_at
                    FROM notification_channels
                    WHERE user_id = %s
                    ORDER BY created_at DESC
                    ''', (user_id,))
                else:
                    cursor.execute('''
                    SELECT id, name, channel_type, config, enabled, created_at, updated_at
                    FROM notification_channels
                    ORDER BY created_at DESC
                    ''')

                channels = []
                for row in cursor.fetchall():
                    channels.append({
                        'id': row[0],
                        'name': row[1],
                        'type': row[2],
                        'config': row[3],
                        'enabled': bool(row[4]),
                        'created_at': row[5],
                        'updated_at': row[6]
                    })

                return channels
            except Exception as e:
                logger.error(f"获取通知渠道失败: {e}")
                return []

    def get_notification_channel(self, channel_id: int) -> Optional[Dict[str, any]]:
        """获取指定通知渠道"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT id, name, channel_type, config, enabled, created_at, updated_at
                FROM notification_channels WHERE id = %s
                ''', (channel_id,))

                row = cursor.fetchone()
                if row:
                    return {
                        'id': row[0],
                        'name': row[1],
                        'type': row[2],
                        'config': row[3],
                        'enabled': bool(row[4]),
                        'created_at': row[5],
                        'updated_at': row[6]
                    }
                return None
            except Exception as e:
                logger.error(f"获取通知渠道失败: {e}")
                return None

    def update_notification_channel(self, channel_id: int, name: str, config: str, enabled: bool = True) -> bool:
        """更新通知渠道"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                UPDATE notification_channels 
                SET name = %s, config = %s, enabled = %s, updated_at = CURRENT_TIMESTAMP
                WHERE id = %s
                ''', (name, config, enabled, channel_id))
                self.conn.commit()
                logger.debug(f"更新通知渠道: {channel_id}")
                return True
            except Exception as e:
                logger.error(f"更新通知渠道失败: {e}")
                self.conn.rollback()
                return False

    def delete_notification_channel(self, channel_id: int) -> bool:
        """删除通知渠道"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('DELETE FROM notification_channels WHERE id = %s', (channel_id,))
                self.conn.commit()
                logger.debug(f"删除通知渠道: {channel_id}")
                return True
            except Exception as e:
                logger.error(f"删除通知渠道失败: {e}")
                self.conn.rollback()
                return False

    # -------------------- 系统设置操作 --------------------
    def get_system_setting(self, key: str) -> Optional[str]:
        """获取系统设置"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('SELECT value FROM system_settings WHERE `key` = %s', (key,))
                result = cursor.fetchone()
                return result[0] if result else None
            except Exception as e:
                logger.error(f"获取系统设置失败: {e}")
                return None

    def set_system_setting(self, key: str, value: str, description: str = None) -> bool:
        """设置系统设置"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                INSERT INTO system_settings (`key`, value, description)
                VALUES (%s, %s, %s)
                ON DUPLICATE KEY UPDATE
                value = VALUES(value),
                description = VALUES(description),
                updated_at = CURRENT_TIMESTAMP
                ''', (key, value, description))
                self.conn.commit()
                return True
            except Exception as e:
                logger.error(f"设置系统设置失败: {e}")
                self.conn.rollback()
                return False

    def get_all_system_settings(self) -> Dict[str, str]:
        """获取所有系统设置"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('SELECT `key`, value FROM system_settings')
                results = cursor.fetchall()
                return {row[0]: row[1] for row in results}
            except Exception as e:
                logger.error(f"获取所有系统设置失败: {e}")
                return {}

    # -------------------- 用户管理操作 --------------------
    def create_user(self, username: str, email: str, password: str) -> bool:
        """创建用户"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                password_hash = hashlib.sha256(password.encode()).hexdigest()
                cursor.execute('''
                INSERT INTO users (username, email, password_hash)
                VALUES (%s, %s, %s)
                ''', (username, email, password_hash))
                self.conn.commit()
                logger.info(f"用户创建成功: {username}")
                return True
            except pymysql.IntegrityError as e:
                if "Duplicate entry" in str(e):
                    logger.error(f"用户创建失败，用户名或邮箱已存在: {username}")
                else:
                    logger.error(f"用户创建失败: {e}")
                self.conn.rollback()
                return False
            except Exception as e:
                logger.error(f"用户创建失败: {e}")
                self.conn.rollback()
                return False

    def get_user_by_username(self, username: str) -> Optional[Dict[str, Any]]:
        """根据用户名获取用户信息"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT id, username, email, password_hash, is_active, is_admin, created_at, updated_at
                FROM users WHERE username = %s
                ''', (username,))
                result = cursor.fetchone()
                
                if result:
                    return {
                        'id': result[0],
                        'username': result[1],
                        'email': result[2],
                        'password_hash': result[3],
                        'is_active': bool(result[4]),
                        'is_admin': bool(result[5]),
                        'created_at': result[6],
                        'updated_at': result[7]
                    }
                return None
            except Exception as e:
                logger.error(f"根据用户名获取用户失败: {e}")
                return None

    def get_user_by_email(self, email: str) -> Optional[Dict[str, Any]]:
        """根据邮箱获取用户信息"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT id, username, email, password_hash, is_active, is_admin, created_at, updated_at
                FROM users WHERE email = %s
                ''', (email,))
                result = cursor.fetchone()
                
                if result:
                    return {
                        'id': result[0],
                        'username': result[1],
                        'email': result[2],
                        'password_hash': result[3],
                        'is_active': bool(result[4]),
                        'is_admin': bool(result[5]),
                        'created_at': result[6],
                        'updated_at': result[7]
                    }
                return None
            except Exception as e:
                logger.error(f"根据邮箱获取用户失败: {e}")
                return None

    def verify_user_password(self, username: str, password: str) -> bool:
        """验证用户密码"""
        user = self.get_user_by_username(username)
        if user:
            password_hash = hashlib.sha256(password.encode()).hexdigest()
            return user['password_hash'] == password_hash
        return False

    def update_user_password(self, username: str, new_password: str) -> bool:
        """更新用户密码"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                password_hash = hashlib.sha256(new_password.encode()).hexdigest()
                cursor.execute('''
                UPDATE users SET password_hash = %s, updated_at = CURRENT_TIMESTAMP
                WHERE username = %s
                ''', (password_hash, username))
                
                if cursor.rowcount > 0:
                    self.conn.commit()
                    logger.info(f"用户密码更新成功: {username}")
                    return True
                else:
                    logger.error(f"用户不存在: {username}")
                    return False
            except Exception as e:
                logger.error(f"更新用户密码失败: {e}")
                self.conn.rollback()
                return False 

    def update_user_info(self, user_id: int, user_data: dict) -> bool:
        """更新用户基本信息"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 构建更新SQL
                update_fields = []
                update_values = []
                
                if 'email' in user_data:
                    update_fields.append('email = %s')
                    update_values.append(user_data['email'])
                
                if 'phone' in user_data:
                    update_fields.append('phone = %s')
                    update_values.append(user_data['phone'])
                
                if 'nickname' in user_data:
                    update_fields.append('nickname = %s')
                    update_values.append(user_data['nickname'])
                
                if not update_fields:
                    return True  # 没有字段需要更新
                
                # 添加更新时间和用户ID
                update_fields.append('updated_at = CURRENT_TIMESTAMP')
                update_values.append(user_id)
                
                sql = f"UPDATE users SET {', '.join(update_fields)} WHERE id = %s"
                cursor.execute(sql, update_values)
                
                if cursor.rowcount > 0:
                    self.conn.commit()
                    logger.info(f"用户信息更新成功: {user_id}")
                    return True
                else:
                    logger.error(f"用户不存在: {user_id}")
                    return False
            except Exception as e:
                logger.error(f"更新用户信息失败: {e}")
                self.conn.rollback()
                return False

    # -------------------- 订单管理操作 --------------------
    def save_order(self, cookie_id: str, order_data: dict) -> bool:
        """保存或更新订单信息"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 检查订单是否已存在
                self._execute_sql(cursor, 
                    "SELECT id FROM orders WHERE cookie_id = %s AND order_id = %s", 
                    (cookie_id, order_data.get('order_id')))
                existing = cursor.fetchone()
                
                if existing:
                    # 更新现有订单
                    if order_data.get('coupon_code') is not None:
                        # 有券码信息，正常更新所有字段
                        self._execute_sql(cursor, '''
                            UPDATE orders SET 
                                buyer_id = %s, item_id = %s, item_title = %s, order_status = %s,
                                order_amount = %s, buyer_message = %s, seller_message = %s,
                                shipping_info = %s, payment_time = %s, shipping_time = %s,
                                delivery_time = %s, refund_time = %s, raw_message = %s,
                                coupon_code = %s, updated_at = CURRENT_TIMESTAMP,
                                spec_name = COALESCE(%s, spec_name),
                                spec_value = COALESCE(%s, spec_value),
                                quantity = COALESCE(%s, quantity),
                                amount = COALESCE(%s, amount)
                            WHERE cookie_id = %s AND order_id = %s
                        ''', (
                            order_data.get('buyer_id'),
                            order_data.get('item_id'),
                            order_data.get('item_title'),
                            order_data.get('order_status'),
                            order_data.get('order_amount'),
                            order_data.get('buyer_message'),
                            order_data.get('seller_message'),
                            order_data.get('shipping_info'),
                            order_data.get('payment_time'),
                            order_data.get('shipping_time'),
                            order_data.get('delivery_time'),
                            order_data.get('refund_time'),
                            order_data.get('raw_message'),
                            order_data.get('coupon_code'),
                            order_data.get('spec_name'),
                            order_data.get('spec_value'),
                            order_data.get('quantity'),
                            order_data.get('amount'),
                            cookie_id,
                            order_data.get('order_id')
                        ))
                    else:
                        # 没有券码信息，更新时保持原有券码信息不变
                        self._execute_sql(cursor, '''
                            UPDATE orders SET 
                                buyer_id = %s, item_id = %s, item_title = %s, order_status = %s,
                                order_amount = %s, buyer_message = %s, seller_message = %s,
                                shipping_info = %s, payment_time = %s, shipping_time = %s,
                                delivery_time = %s, refund_time = %s, raw_message = %s,
                                updated_at = CURRENT_TIMESTAMP,
                                spec_name = COALESCE(%s, spec_name),
                                spec_value = COALESCE(%s, spec_value),
                                quantity = COALESCE(%s, quantity),
                                amount = COALESCE(%s, amount)
                            WHERE cookie_id = %s AND order_id = %s
                        ''', (
                            order_data.get('buyer_id'),
                            order_data.get('item_id'),
                            order_data.get('item_title'),
                            order_data.get('order_status'),
                            order_data.get('order_amount'),
                            order_data.get('buyer_message'),
                            order_data.get('seller_message'),
                            order_data.get('shipping_info'),
                            order_data.get('payment_time'),
                            order_data.get('shipping_time'),
                            order_data.get('delivery_time'),
                            order_data.get('refund_time'),
                            order_data.get('raw_message'),
                            order_data.get('spec_name'),
                            order_data.get('spec_value'),
                            order_data.get('quantity'),
                            order_data.get('amount'),
                            cookie_id,
                            order_data.get('order_id')
                        ))
                    logger.info(f"订单更新成功: {cookie_id} - {order_data.get('order_id')}")
                else:
                    # 插入新订单
                    self._execute_sql(cursor, '''
                        INSERT INTO orders (
                            cookie_id, order_id, buyer_id, item_id, item_title,
                            order_status, order_amount, buyer_message, seller_message,
                            shipping_info, payment_time, shipping_time, delivery_time,
                            refund_time, raw_message, coupon_code,
                            spec_name, spec_value, quantity, amount
                        ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
                    ''', (
                        cookie_id,
                        order_data.get('order_id'),
                        order_data.get('buyer_id'),
                        order_data.get('item_id'),
                        order_data.get('item_title'),
                        order_data.get('order_status'),
                        order_data.get('order_amount'),
                        order_data.get('buyer_message'),
                        order_data.get('seller_message'),
                        order_data.get('shipping_info'),
                        order_data.get('payment_time'),
                        order_data.get('shipping_time'),
                        order_data.get('delivery_time'),
                        order_data.get('refund_time'),
                        order_data.get('raw_message'),
                        order_data.get('coupon_code'),
                        order_data.get('spec_name'),
                        order_data.get('spec_value'),
                        order_data.get('quantity'),
                        order_data.get('amount')
                    ))
                    logger.info(f"订单创建成功: {cookie_id} - {order_data.get('order_id')}")
                
                self.conn.commit()
                return True
            except Exception as e:
                logger.error(f"保存订单失败: {e}")
                self.conn.rollback()
                return False

    def get_orders(self, cookie_id: str = None, page: int = 1, page_size: int = 20, 
                   search_query: str = None, search_type: str = None, status_filter: str = None) -> dict:
        """获取订单列表（支持分页和搜索）"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 构建查询条件
                where_conditions = []
                params = []
                
                if cookie_id:
                    where_conditions.append("cookie_id = %s")
                    params.append(cookie_id)
                
                if search_query:
                    search_param = f"%{search_query}%"
                    if search_type and search_type != 'all':
                        # 按指定类型搜索
                        if search_type == 'order_id':
                            where_conditions.append("order_id LIKE %s")
                            params.append(search_param)
                        elif search_type == 'buyer_id':
                            where_conditions.append("buyer_id LIKE %s")
                            params.append(search_param)
                        elif search_type == 'item_title':
                            where_conditions.append("item_title LIKE %s")
                            params.append(search_param)
                        elif search_type == 'coupon_code':
                            where_conditions.append("coupon_code LIKE %s")
                            params.append(search_param)
                    else:
                        # 全字段搜索
                        where_conditions.append(
                            "(order_id LIKE %s OR buyer_id LIKE %s OR item_title LIKE %s OR buyer_message LIKE %s OR coupon_code LIKE %s)"
                        )
                        params.extend([search_param, search_param, search_param, search_param, search_param])
                
                if status_filter:
                    where_conditions.append("order_status = %s")
                    params.append(status_filter)
                
                where_clause = " WHERE " + " AND ".join(where_conditions) if where_conditions else ""
                
                # 获取总数
                count_sql = f"SELECT COUNT(*) FROM orders{where_clause}"
                self._execute_sql(cursor, count_sql, params)
                total_count = cursor.fetchone()[0]
                
                # 计算分页
                total_pages = (total_count + page_size - 1) // page_size
                offset = (page - 1) * page_size
                
                # 获取订单数据
                data_sql = f'''
                    SELECT id, cookie_id, order_id, buyer_id, item_id, item_title,
                           order_status, order_amount, buyer_message, seller_message,
                           shipping_info, payment_time, shipping_time, delivery_time,
                           refund_time, coupon_code, created_at, updated_at,
                           spec_name, spec_value, quantity, amount
                    FROM orders{where_clause}
                    ORDER BY updated_at DESC
                    LIMIT %s OFFSET %s
                '''
                self._execute_sql(cursor, data_sql, params + [page_size, offset])
                
                orders = []
                for row in cursor.fetchall():
                    orders.append({
                        'id': row[0],
                        'cookie_id': row[1],
                        'order_id': row[2],
                        'buyer_id': row[3],
                        'item_id': row[4],
                        'item_title': row[5],
                        'order_status': row[6],
                        'order_amount': float(row[7]) if row[7] else None,
                        'buyer_message': row[8],
                        'seller_message': row[9],
                        'shipping_info': row[10],
                        'payment_time': row[11],
                        'shipping_time': row[12],
                        'delivery_time': row[13],
                        'refund_time': row[14],
                        'coupon_code': row[15],
                        'delivery_coupon_code': row[15],  # 为前端兼容性添加别名
                        'created_at': row[16],
                        'updated_at': row[17],
                        'spec_name': row[18],
                        'spec_value': row[19],
                        'quantity': row[20],
                        'amount': row[21]
                    })
                
                return {
                    'orders': orders,
                    'total_count': total_count,
                    'total_pages': total_pages,
                    'current_page': page,
                    'page_size': page_size
                }
                
            except Exception as e:
                logger.error(f"获取订单列表失败: {e}")
                return {
                    'orders': [],
                    'total_count': 0,
                    'total_pages': 0,
                    'current_page': page,
                    'page_size': page_size
                }

    def get_order_by_id(self, order_id: str, cookie_id: str = None) -> Optional[dict]:
        """根据订单ID获取订单详情"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                if cookie_id:
                    cursor.execute('''
                    SELECT id, cookie_id, order_id, buyer_id, item_id, item_title, order_status, 
                           order_amount, buyer_message, seller_message, shipping_info, payment_time, 
                           shipping_time, delivery_time, refund_time, coupon_code, created_at, 
                           updated_at, spec_name, spec_value, quantity, amount
                    FROM orders 
                    WHERE order_id = %s AND cookie_id = %s
                    ''', (order_id, cookie_id))
                else:
                    cursor.execute('''
                    SELECT id, cookie_id, order_id, buyer_id, item_id, item_title, order_status, 
                           order_amount, buyer_message, seller_message, shipping_info, payment_time, 
                           shipping_time, delivery_time, refund_time, coupon_code, created_at, 
                           updated_at, spec_name, spec_value, quantity, amount
                    FROM orders 
                    WHERE order_id = %s
                    ''', (order_id,))
                
                row = cursor.fetchone()
                if row:
                    return {
                        'id': row[0],
                        'cookie_id': row[1],
                        'order_id': row[2],
                        'buyer_id': row[3],
                        'item_id': row[4],
                        'item_title': row[5],
                        'order_status': row[6],
                        'order_amount': row[7],
                        'buyer_message': row[8],
                        'seller_message': row[9],
                        'shipping_info': row[10],
                        'payment_time': row[11],
                        'shipping_time': row[12],
                        'delivery_time': row[13],
                        'refund_time': row[14],
                        'coupon_code': row[15],
                        'delivery_coupon_code': row[15],  # 为前端兼容性添加别名
                        'created_at': row[16],
                        'updated_at': row[17],
                        'spec_name': row[18],
                        'spec_value': row[19],
                        'quantity': row[20],
                        'amount': row[21]
                    }
                return None
            except Exception as e:
                logger.error(f"获取订单详情失败: {e}")
                return None

    def get_order_by_id_enhanced(self, order_id: str) -> Optional[dict]:
        """根据订单ID获取增强的订单信息"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT order_id, item_id, buyer_id, spec_name, spec_value,
                       quantity, amount, order_status, cookie_id, created_at, updated_at
                FROM orders WHERE order_id = %s
                ''', (order_id,))

                row = cursor.fetchone()
                if row:
                    return {
                        'order_id': row[0],
                        'item_id': row[1],
                        'buyer_id': row[2],
                        'spec_name': row[3],
                        'spec_value': row[4],
                        'quantity': row[5],
                        'amount': row[6],
                        'order_status': row[7],
                        'cookie_id': row[8],
                        'created_at': row[9],
                        'updated_at': row[10]
                    }
                return None

            except Exception as e:
                logger.error(f"获取订单信息失败: {order_id} - {e}")
                return None

    def get_recent_order_by_item_id(self, item_id: str, cookie_id: str) -> Optional[dict]:
        """根据商品ID和账号ID获取最近的订单"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT order_id, item_id, buyer_id, spec_name, spec_value,
                       quantity, amount, order_status, cookie_id, created_at, updated_at
                FROM orders 
                WHERE item_id = %s AND cookie_id = %s 
                ORDER BY created_at DESC 
                LIMIT 1
                ''', (item_id, cookie_id))
                
                result = cursor.fetchone()
                if result:
                    return {
                        'order_id': result[0],
                        'item_id': result[1],
                        'buyer_id': result[2],
                        'spec_name': result[3],
                        'spec_value': result[4],
                        'quantity': result[5],
                        'amount': result[6],
                        'order_status': result[7],
                        'cookie_id': result[8],
                        'created_at': result[9],
                        'updated_at': result[10]
                    }
                return None
            except Exception as e:
                logger.error(f"根据商品ID获取最近订单失败: {e}")
                return None

    def insert_or_update_order(self, order_id: str, item_id: str = None, buyer_id: str = None,
                              spec_name: str = None, spec_value: str = None, quantity: str = None,
                              amount: str = None, order_status: str = None, cookie_id: str = None):
        """插入或更新订单信息"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()

                # 检查cookie_id是否在cookies表中存在（如果提供了cookie_id）
                if cookie_id:
                    cursor.execute("SELECT id FROM cookies WHERE id = %s", (cookie_id,))
                    cookie_exists = cursor.fetchone()
                    if not cookie_exists:
                        logger.warning(f"Cookie ID {cookie_id} 不存在于cookies表中，拒绝插入订单 {order_id}")
                        return False

                # 检查订单是否已存在
                cursor.execute("SELECT order_id FROM orders WHERE order_id = %s", (order_id,))
                existing = cursor.fetchone()

                if existing:
                    # 更新现有订单
                    update_fields = []
                    update_values = []

                    if item_id is not None:
                        update_fields.append("item_id = %s")
                        update_values.append(item_id)
                    if buyer_id is not None:
                        update_fields.append("buyer_id = %s")
                        update_values.append(buyer_id)
                    if spec_name is not None:
                        update_fields.append("spec_name = %s")
                        update_values.append(spec_name)
                    if spec_value is not None:
                        update_fields.append("spec_value = %s")
                        update_values.append(spec_value)
                    if quantity is not None:
                        update_fields.append("quantity = %s")
                        update_values.append(quantity)
                    if amount is not None:
                        update_fields.append("amount = %s")
                        update_values.append(amount)
                    if order_status is not None:
                        update_fields.append("order_status = %s")
                        update_values.append(order_status)
                    if cookie_id is not None:
                        update_fields.append("cookie_id = %s")
                        update_values.append(cookie_id)

                    if update_fields:
                        update_fields.append("updated_at = CURRENT_TIMESTAMP")
                        update_values.append(order_id)

                        sql = f"UPDATE orders SET {', '.join(update_fields)} WHERE order_id = %s"
                        self._execute_sql(cursor, sql, update_values)
                        logger.info(f"更新订单信息: {order_id}")
                else:
                    # 插入新订单
                    self._execute_sql(cursor, '''
                    INSERT INTO orders (order_id, item_id, buyer_id, spec_name, spec_value,
                                      quantity, amount, order_status, cookie_id)
                    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
                    ''', (order_id, item_id, buyer_id, spec_name, spec_value,
                          quantity, amount, order_status or 'unknown', cookie_id))
                    logger.info(f"插入新订单: {order_id}")

                self.conn.commit()
                return True

            except Exception as e:
                logger.error(f"插入或更新订单失败: {order_id} - {e}")
                self.conn.rollback()
                return False

    def delete_order(self, order_id: str, cookie_id: str) -> bool:
        """删除订单"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                self._execute_sql(cursor, 
                    "DELETE FROM orders WHERE order_id = %s AND cookie_id = %s", 
                    (order_id, cookie_id))
                
                if cursor.rowcount > 0:
                    self.conn.commit()
                    logger.info(f"订单删除成功: {order_id}")
                    return True
                else:
                    logger.warning(f"订单不存在或无权删除: {order_id}")
                    return False
            except Exception as e:
                logger.error(f"删除订单失败: {e}")
                self.conn.rollback()
                return False

    def get_order_statistics(self, cookie_id: str = None) -> dict:
        """获取订单统计信息"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                base_where = "WHERE cookie_id = %s" if cookie_id else ""
                params = [cookie_id] if cookie_id else []
                
                # 总订单数
                self._execute_sql(cursor, f"SELECT COUNT(*) FROM orders {base_where}", params)
                total_orders = cursor.fetchone()[0]
                
                # 按状态统计 - 修改字段名为status_counts以匹配前端期望
                status_counts = {}
                status_list = ['待付款', '待发货', '已发货', '已完成', '退款中', '已退款', '已取消']
                
                for status in status_list:
                    where_clause = f"{base_where} {'AND' if base_where else 'WHERE'} order_status = %s"
                    self._execute_sql(cursor, f"SELECT COUNT(*) FROM orders {where_clause}", 
                                    params + [status])
                    count = cursor.fetchone()
                    status_counts[status] = count[0] if count else 0
                
                # 今日订单数
                self._execute_sql(cursor, f'''
                    SELECT COUNT(*) FROM orders 
                    {base_where} {'AND' if base_where else 'WHERE'} DATE(created_at) = CURDATE()
                ''', params)
                today_orders = cursor.fetchone()[0]
                
                # 本月订单数
                self._execute_sql(cursor, f'''
                    SELECT COUNT(*) FROM orders 
                    {base_where} {'AND' if base_where else 'WHERE'} YEAR(created_at) = YEAR(NOW()) AND MONTH(created_at) = MONTH(NOW())
                ''', params)
                month_orders = cursor.fetchone()[0]
                
                # 本月订单总金额 - 新增缺失的字段
                month_where = f"{base_where} {'AND' if base_where else 'WHERE'} YEAR(created_at) = YEAR(NOW()) AND MONTH(created_at) = MONTH(NOW())"
                amount_where = f"{month_where} AND amount IS NOT NULL AND amount != '' AND amount != '-'"
                self._execute_sql(cursor, f'''
                    SELECT COALESCE(SUM(CAST(REPLACE(REPLACE(amount, '¥', ''), ',', '') AS DECIMAL(10,2))), 0)
                    FROM orders {amount_where}
                ''', params)
                month_amount_result = cursor.fetchone()
                month_amount = float(month_amount_result[0]) if month_amount_result and month_amount_result[0] else 0.0
                
                return {
                    'total_orders': total_orders,
                    'status_counts': status_counts,  # 修改字段名以匹配前端
                    'today_orders': today_orders,
                    'month_orders': month_orders,
                    'month_amount': month_amount  # 新增本月交易额字段
                }
                
            except Exception as e:
                logger.error(f"获取订单统计失败: {e}")
                return {
                    'total_orders': 0,
                    'status_counts': {},  # 修改字段名以匹配前端
                    'today_orders': 0,
                    'month_orders': 0,
                    'month_amount': 0.0  # 新增本月交易额字段
                }

    # -------------------- 商品信息操作 --------------------
    def save_item_basic_info(self, cookie_id: str, item_id: str, item_title: str = None,
                            item_description: str = None, item_category: str = None,
                            item_price: str = None, item_detail: str = None) -> bool:
        """保存商品基本信息"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 使用INSERT ... ON DUPLICATE KEY UPDATE
                cursor.execute('''
                INSERT INTO item_info (cookie_id, item_id, item_title, item_description, 
                                     item_category, item_price, item_detail)
                VALUES (%s, %s, %s, %s, %s, %s, %s)
                ON DUPLICATE KEY UPDATE
                item_title = VALUES(item_title),
                item_description = VALUES(item_description),
                item_category = VALUES(item_category),
                item_price = VALUES(item_price),
                item_detail = VALUES(item_detail),
                updated_at = CURRENT_TIMESTAMP
                ''', (cookie_id, item_id, item_title, item_description, 
                      item_category, item_price, item_detail))
                
                self.conn.commit()
                logger.info(f"商品信息保存成功: {cookie_id} - {item_id}")
                return True
            except Exception as e:
                logger.error(f"保存商品信息失败: {e}")
                self.conn.rollback()
                return False

    def get_item_info(self, cookie_id: str, item_id: str) -> Optional[Dict]:
        """获取商品信息"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT cookie_id, item_id, item_title, item_description, item_category,
                       item_price, item_detail, is_multi_spec, created_at, updated_at
                FROM item_info WHERE cookie_id = %s AND item_id = %s
                ''', (cookie_id, item_id))
                
                result = cursor.fetchone()
                if result:
                    return {
                        'cookie_id': result[0],
                        'item_id': result[1],
                        'item_title': result[2],
                        'item_description': result[3],
                        'item_category': result[4],
                        'item_price': result[5],
                        'item_detail': result[6],
                        'is_multi_spec': bool(result[7]),
                        'created_at': result[8],
                        'updated_at': result[9]
                    }
                return None
            except Exception as e:
                logger.error(f"获取商品信息失败: {e}")
                return None

    def update_item_multi_spec_status(self, cookie_id: str, item_id: str, is_multi_spec: bool) -> bool:
        """更新商品的多规格状态"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                UPDATE item_info
                SET is_multi_spec = %s, updated_at = CURRENT_TIMESTAMP
                WHERE cookie_id = %s AND item_id = %s
                ''', (is_multi_spec, cookie_id, item_id))

                if cursor.rowcount > 0:
                    self.conn.commit()
                    logger.info(f"更新商品多规格状态成功: {item_id} -> {is_multi_spec}")
                    return True
                else:
                    logger.warning(f"商品不存在，无法更新多规格状态: {item_id}")
                    return False

            except Exception as e:
                logger.error(f"更新商品多规格状态失败: {e}")
                self.conn.rollback()
                return False

    def get_item_multi_spec_status(self, cookie_id: str, item_id: str) -> bool:
        """获取商品的多规格状态"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT is_multi_spec FROM item_info
                WHERE cookie_id = %s AND item_id = %s
                ''', (cookie_id, item_id))

                row = cursor.fetchone()
                if row:
                    return bool(row[0]) if row[0] is not None else False
                return False

            except Exception as e:
                logger.error(f"获取商品多规格状态失败: {e}")
                return False

    def get_items_by_cookie(self, cookie_id: str) -> List[Dict]:
        """获取指定Cookie的所有商品"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT cookie_id, item_id, item_title, item_description, item_category,
                       item_price, item_detail, is_multi_spec, created_at, updated_at
                FROM item_info WHERE cookie_id = %s
                ORDER BY updated_at DESC
                ''', (cookie_id,))
                
                items = []
                for row in cursor.fetchall():
                    items.append({
                        'cookie_id': row[0],
                        'item_id': row[1],
                        'item_title': row[2],
                        'item_description': row[3],
                        'item_category': row[4],
                        'item_price': row[5],
                        'item_detail': row[6],
                        'is_multi_spec': bool(row[7]),
                        'created_at': row[8],
                        'updated_at': row[9]
                    })
                
                return items
            except Exception as e:
                logger.error(f"获取商品列表失败: {e}")
                return []

    def batch_save_item_basic_info(self, items_data: list) -> int:
        """批量保存商品基本信息（并发安全）

        Args:
            items_data: 商品数据列表，每个元素包含 cookie_id, item_id, item_title 等字段

        Returns:
            int: 成功保存的商品数量
        """
        if not items_data:
            return 0

        success_count = 0
        with self.lock:
            try:
                cursor = self.get_connection().cursor()

                for item_data in items_data:
                    try:
                        cookie_id = item_data.get('cookie_id')
                        item_id = item_data.get('item_id')
                        item_title = item_data.get('item_title', '')
                        item_description = item_data.get('item_description', '')
                        item_category = item_data.get('item_category', '')
                        item_price = item_data.get('item_price', '')
                        item_detail = item_data.get('item_detail', '')

                        if not cookie_id or not item_id:
                            continue

                        # 验证：如果没有商品标题，则跳过保存
                        if not item_title or not item_title.strip():
                            logger.debug(f"跳过批量保存商品信息：缺少商品标题 - {item_id}")
                            continue

                        # 使用 INSERT ... ON DUPLICATE KEY UPDATE
                        self._execute_sql(cursor, '''
                        INSERT INTO item_info (cookie_id, item_id, item_title, item_description,
                                             item_category, item_price, item_detail)
                        VALUES (%s, %s, %s, %s, %s, %s, %s)
                        ON DUPLICATE KEY UPDATE
                        item_title = CASE WHEN (item_title IS NULL OR item_title = '') AND VALUES(item_title) != '' THEN VALUES(item_title) ELSE item_title END,
                        item_description = CASE WHEN (item_description IS NULL OR item_description = '') AND VALUES(item_description) != '' THEN VALUES(item_description) ELSE item_description END,
                        item_category = CASE WHEN (item_category IS NULL OR item_category = '') AND VALUES(item_category) != '' THEN VALUES(item_category) ELSE item_category END,
                        item_price = CASE WHEN (item_price IS NULL OR item_price = '') AND VALUES(item_price) != '' THEN VALUES(item_price) ELSE item_price END,
                        item_detail = CASE WHEN (item_detail IS NULL OR item_detail = '' OR TRIM(item_detail) = '') AND VALUES(item_detail) != '' THEN VALUES(item_detail) ELSE item_detail END,
                        updated_at = CURRENT_TIMESTAMP
                        ''', (cookie_id, item_id, item_title, item_description,
                              item_category, item_price, item_detail))

                        success_count += 1

                    except Exception as item_e:
                        logger.warning(f"批量保存单个商品失败 {item_data.get('item_id', 'unknown')}: {item_e}")
                        continue

                self.conn.commit()
                logger.info(f"批量保存商品信息完成: {success_count}/{len(items_data)} 个商品")
                return success_count

            except Exception as e:
                logger.error(f"批量保存商品信息失败: {e}")
                self.conn.rollback()
                return success_count

    def save_item_info(self, cookie_id: str, item_id: str, item_data = None) -> bool:
        """保存或更新商品信息

        Args:
            cookie_id: Cookie ID
            item_id: 商品ID
            item_data: 商品详情数据，可以是字符串或字典，也可以为None

        Returns:
            bool: 操作是否成功
        """
        try:
            # 验证：如果只有商品ID，没有商品详情数据，则不插入数据库
            if not item_data:
                logger.debug(f"跳过保存商品信息：缺少商品详情数据 - {item_id}")
                return False

            # 如果是字典类型，检查是否有标题信息
            if isinstance(item_data, dict):
                title = item_data.get('title', '').strip()
                if not title:
                    logger.debug(f"跳过保存商品信息：缺少商品标题 - {item_id}")
                    return False

            # 如果是字符串类型，检查是否为空
            if isinstance(item_data, str) and not item_data.strip():
                logger.debug(f"跳过保存商品信息：商品详情为空 - {item_id}")
                return False

            with self.lock:
                cursor = self.get_connection().cursor()

                # 如果是字典类型，提取各个字段
                if isinstance(item_data, dict):
                    item_title = item_data.get('title', '')
                    item_description = item_data.get('description', '')
                    item_category = item_data.get('category', '')
                    item_price = item_data.get('price', '')
                    item_detail = str(item_data)  # 将整个字典作为详情
                else:
                    # 字符串类型，作为详情保存
                    item_title = ''
                    item_description = ''
                    item_category = ''
                    item_price = ''
                    item_detail = str(item_data)

                # 使用 INSERT ... ON DUPLICATE KEY UPDATE
                self._execute_sql(cursor, '''
                INSERT INTO item_info (cookie_id, item_id, item_title, item_description, 
                                     item_category, item_price, item_detail)
                VALUES (%s, %s, %s, %s, %s, %s, %s)
                ON DUPLICATE KEY UPDATE
                item_title = CASE WHEN (item_title IS NULL OR item_title = '') AND VALUES(item_title) != '' THEN VALUES(item_title) ELSE item_title END,
                item_description = CASE WHEN (item_description IS NULL OR item_description = '') AND VALUES(item_description) != '' THEN VALUES(item_description) ELSE item_description END,
                item_category = CASE WHEN (item_category IS NULL OR item_category = '') AND VALUES(item_category) != '' THEN VALUES(item_category) ELSE item_category END,
                item_price = CASE WHEN (item_price IS NULL OR item_price = '') AND VALUES(item_price) != '' THEN VALUES(item_price) ELSE item_price END,
                item_detail = VALUES(item_detail),
                updated_at = CURRENT_TIMESTAMP
                ''', (cookie_id, item_id, item_title, item_description, 
                      item_category, item_price, item_detail))

                self.conn.commit()
                logger.debug(f"商品信息保存成功: {cookie_id} - {item_id}")
                return True

        except Exception as e:
            logger.error(f"保存商品信息失败: {e}")
            if hasattr(self, 'conn'):
                self.conn.rollback()
            return False

    # -------------------- 验证码操作 --------------------
    def generate_verification_code(self) -> str:
        """生成6位数字验证码"""
        return ''.join(random.choices(string.digits, k=6))

    def generate_captcha(self) -> Tuple[str, str]:
        """生成图形验证码
        返回: (验证码文本, base64编码的图片)
        """
        try:
            # 生成4位随机验证码（数字+字母）
            chars = string.ascii_uppercase + string.digits
            captcha_text = ''.join(random.choices(chars, k=4))

            # 创建图片
            width, height = 120, 40
            image = Image.new('RGB', (width, height), color='white')
            draw = ImageDraw.Draw(image)

            # 尝试使用系统字体，如果失败则使用默认字体
            try:
                # Windows系统字体
                font = ImageFont.truetype("arial.ttf", 20)
            except:
                try:
                    # 备用字体
                    font = ImageFont.truetype("C:/Windows/Fonts/arial.ttf", 20)
                except:
                    # 使用默认字体
                    font = ImageFont.load_default()

            # 绘制验证码文本
            for i, char in enumerate(captcha_text):
                # 随机颜色
                color = (
                    random.randint(0, 100),
                    random.randint(0, 100),
                    random.randint(0, 100)
                )

                # 随机位置（稍微偏移）
                x = 20 + i * 20 + random.randint(-3, 3)
                y = 8 + random.randint(-3, 3)

                draw.text((x, y), char, font=font, fill=color)

            # 添加干扰线
            for _ in range(3):
                start = (random.randint(0, width), random.randint(0, height))
                end = (random.randint(0, width), random.randint(0, height))
                draw.line([start, end], fill=(random.randint(100, 200), random.randint(100, 200), random.randint(100, 200)), width=1)

            # 添加干扰点
            for _ in range(20):
                x = random.randint(0, width)
                y = random.randint(0, height)
                draw.point((x, y), fill=(random.randint(0, 255), random.randint(0, 255), random.randint(0, 255)))

            # 转换为base64
            buffer = io.BytesIO()
            image.save(buffer, format='PNG')
            img_base64 = base64.b64encode(buffer.getvalue()).decode()

            return captcha_text, f"data:image/png;base64,{img_base64}"

        except Exception as e:
            logger.error(f"生成图形验证码失败: {e}")
            # 返回简单的文本验证码作为备用
            simple_code = ''.join(random.choices(string.digits, k=4))
            return simple_code, ""

    def save_captcha(self, session_id: str, captcha_text: str, expires_minutes: int = 5) -> bool:
        """保存图形验证码"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                expires_at = time.time() + (expires_minutes * 60)

                # 删除该session的旧验证码
                cursor.execute('DELETE FROM captcha_sessions WHERE session_id = %s', (session_id,))

                cursor.execute('''
                INSERT INTO captcha_sessions (session_id, captcha_text, expires_at)
                VALUES (%s, %s, FROM_UNIXTIME(%s))
                ''', (session_id, captcha_text.upper(), expires_at))

                self.conn.commit()
                logger.debug(f"保存图形验证码成功: {session_id}")
                return True
            except Exception as e:
                logger.error(f"保存图形验证码失败: {e}")
                self.conn.rollback()
                return False

    def verify_captcha(self, session_id: str, user_input: str) -> bool:
        """验证图形验证码"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                current_time = time.time()

                # 查找有效的验证码
                cursor.execute('''
                SELECT id FROM captcha_sessions
                WHERE session_id = %s AND captcha_text = %s AND expires_at > FROM_UNIXTIME(%s)
                ORDER BY created_at DESC LIMIT 1
                ''', (session_id, user_input.upper(), current_time))

                row = cursor.fetchone()
                if row:
                    # 删除已使用的验证码
                    cursor.execute('DELETE FROM captcha_sessions WHERE id = %s', (row[0],))
                    self.conn.commit()
                    logger.debug(f"图形验证码验证成功: {session_id}")
                    return True
                else:
                    logger.warning(f"图形验证码验证失败: {session_id} - {user_input}")
                    return False
            except Exception as e:
                logger.error(f"验证图形验证码失败: {e}")
                return False

    def save_verification_code(self, email: str, code: str, code_type: str = 'register', expires_minutes: int = 10) -> bool:
        """保存邮箱验证码"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                expires_at = time.time() + (expires_minutes * 60)

                cursor.execute('''
                INSERT INTO email_verifications (email, code, type, expires_at)
                VALUES (%s, %s, %s, FROM_UNIXTIME(%s))
                ''', (email, code, code_type, expires_at))

                self.conn.commit()
                logger.info(f"保存验证码成功: {email} ({code_type})")
                return True
            except Exception as e:
                logger.error(f"保存验证码失败: {e}")
                self.conn.rollback()
                return False

    def verify_email_code(self, email: str, code: str, code_type: str = 'register') -> bool:
        """验证邮箱验证码"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                current_time = time.time()

                # 查找有效的验证码
                cursor.execute('''
                SELECT id FROM email_verifications
                WHERE email = %s AND code = %s AND type = %s AND expires_at > FROM_UNIXTIME(%s) AND used = FALSE
                ORDER BY created_at DESC LIMIT 1
                ''', (email, code, code_type, current_time))

                row = cursor.fetchone()
                if row:
                    # 标记验证码为已使用
                    cursor.execute('''
                    UPDATE email_verifications SET used = TRUE WHERE id = %s
                    ''', (row[0],))
                    self.conn.commit()
                    logger.info(f"验证码验证成功: {email} ({code_type})")
                    return True
                else:
                    logger.warning(f"验证码验证失败: {email} - {code} ({code_type})")
                    return False
            except Exception as e:
                logger.error(f"验证邮箱验证码失败: {e}")
                return False

    async def send_verification_email(self, email: str, code: str) -> bool:
        """发送验证码邮件"""
        try:
            subject = "闲鱼自动回复系统 - 邮箱验证码"
            # 使用简单的纯文本邮件内容
            text_content = f"""【闲鱼自动回复系统】邮箱验证码

您好！

感谢您使用闲鱼自动回复系统。为了确保账户安全，请使用以下验证码完成邮箱验证：

验证码：{code}

重要提醒：
• 验证码有效期为 10 分钟，请及时使用
• 请勿将验证码分享给任何人
• 如非本人操作，请忽略此邮件
• 系统不会主动索要您的验证码

如果您在使用过程中遇到任何问题，请联系我们的技术支持团队。
感谢您选择闲鱼自动回复系统！

---
此邮件由系统自动发送，请勿直接回复
© 2025 闲鱼自动回复系统"""

            # 使用GET请求发送邮件
            api_url = "https://dy.zhinianboke.com/api/emailSend"
            params = {
                'subject': subject,
                'receiveUser': email,
                'sendHtml': text_content
            }

            async with aiohttp.ClientSession() as session:
                try:
                    logger.info(f"发送验证码邮件: {email}")
                    async with session.get(api_url, params=params, timeout=15) as response:
                        response_text = await response.text()
                        logger.info(f"邮件API响应: {response.status}")

                        if response.status == 200:
                            logger.info(f"验证码邮件发送成功: {email}")
                            return True
                        else:
                            logger.error(f"验证码邮件发送失败: {email}, 状态码: {response.status}, 响应: {response_text[:200]}")
                            return False
                except Exception as e:
                    logger.error(f"邮件发送异常: {email}, 错误: {e}")
                    return False

        except Exception as e:
            logger.error(f"发送验证码邮件异常: {e}")
            return False

    # ==================== 卡种管理方法 ====================
    def create_card(self, name: str, card_type: str, api_config=None,
                   text_content: str = None, data_content: str = None, image_url: str = None,
                   description: str = None, enabled: bool = True, delay_seconds: int = 0,
                   user_id: int = None, stock_warning_threshold: int = 0):
        """创建卡券"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 如果没有指定用户ID，使用默认的admin用户
                if user_id is None:
                    cursor.execute("SELECT id FROM users WHERE username = 'admin' LIMIT 1")
                    admin_user = cursor.fetchone()
                    user_id = admin_user[0] if admin_user else 1

                # 转换api_config为JSON字符串
                api_config_json = json.dumps(api_config) if api_config else None

                cursor.execute('''
                INSERT INTO cards (name, type, api_config, text_content, data_content, image_url,
                                 description, enabled, delay_seconds, user_id, stock_warning_threshold)
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
                ''', (name, card_type, api_config_json, text_content, data_content, image_url,
                      description, enabled, delay_seconds, user_id, stock_warning_threshold))
                
                card_id = cursor.lastrowid
                self.conn.commit()
                logger.info(f"卡券创建成功: {name} (ID: {card_id})")
                return card_id
            except Exception as e:
                logger.error(f"创建卡券失败: {e}")
                self.conn.rollback()
                raise

    def get_all_cards(self, user_id: int = None):
        """获取所有卡券"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                if user_id is not None:
                    cursor.execute('''
                    SELECT id, name, type, api_config, text_content, data_content, image_url,
                           description, enabled, delay_seconds, user_id, created_at, updated_at
                    FROM cards WHERE user_id = %s ORDER BY created_at DESC
                    ''', (user_id,))
                else:
                    cursor.execute('''
                    SELECT id, name, type, api_config, text_content, data_content, image_url,
                           description, enabled, delay_seconds, user_id, created_at, updated_at
                    FROM cards ORDER BY created_at DESC
                    ''')

                cards = []
                for row in cursor.fetchall():
                    # 解析JSON配置
                    api_config = None
                    if row[3]:
                        try:
                            api_config = json.loads(row[3])
                        except:
                            api_config = {}

                    cards.append({
                        'id': row[0],
                        'name': row[1],
                        'type': row[2],
                        'api_config': api_config,
                        'text_content': row[4],
                        'data_content': row[5],
                        'image_url': row[6],
                        'description': row[7],
                        'enabled': bool(row[8]),
                        'delay_seconds': row[9],
                        'user_id': row[10],
                        'created_at': row[11],
                        'updated_at': row[12]
                    })

                return cards
            except Exception as e:
                logger.error(f"获取所有卡券失败: {e}")
                return []

    def get_card_by_id(self, card_id: int, user_id: int = None):
        """根据ID获取卡券"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                if user_id is not None:
                    cursor.execute('''
                    SELECT id, name, type, api_config, text_content, data_content, image_url,
                           description, enabled, delay_seconds, user_id, created_at, updated_at
                    FROM cards WHERE id = %s AND user_id = %s
                    ''', (card_id, user_id))
                else:
                    cursor.execute('''
                    SELECT id, name, type, api_config, text_content, data_content, image_url,
                           description, enabled, delay_seconds, user_id, created_at, updated_at
                    FROM cards WHERE id = %s
                    ''', (card_id,))

                row = cursor.fetchone()
                if row:
                    # 解析JSON配置
                    api_config = None
                    if row[3]:
                        try:
                            api_config = json.loads(row[3])
                        except:
                            api_config = {}

                    return {
                        'id': row[0],
                        'name': row[1],
                        'type': row[2],
                        'api_config': api_config,
                        'text_content': row[4],
                        'data_content': row[5],
                        'image_url': row[6],
                        'description': row[7],
                        'enabled': bool(row[8]),
                        'delay_seconds': row[9],
                        'user_id': row[10],
                        'created_at': row[11],
                        'updated_at': row[12]
                    }
                return None
            except Exception as e:
                logger.error(f"根据ID获取卡券失败: {e}")
                return None

    def update_card(self, card_id: int, name: str = None, card_type: str = None,
                   api_config=None, text_content: str = None, data_content: str = None,
                   image_url: str = None, description: str = None, enabled: bool = None,
                   delay_seconds: int = None, stock_warning_threshold: int = None):
        """更新卡券"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 构建动态更新SQL
                update_fields = []
                params = []
                
                if name is not None:
                    update_fields.append("name = %s")
                    params.append(name)
                if card_type is not None:
                    update_fields.append("type = %s")
                    params.append(card_type)
                if api_config is not None:
                    update_fields.append("api_config = %s")
                    params.append(json.dumps(api_config))
                if text_content is not None:
                    update_fields.append("text_content = %s")
                    params.append(text_content)
                if data_content is not None:
                    update_fields.append("data_content = %s")
                    params.append(data_content)
                if image_url is not None:
                    update_fields.append("image_url = %s")
                    params.append(image_url)
                if description is not None:
                    update_fields.append("description = %s")
                    params.append(description)
                if enabled is not None:
                    update_fields.append("enabled = %s")
                    params.append(enabled)
                if delay_seconds is not None:
                    update_fields.append("delay_seconds = %s")
                    params.append(delay_seconds)
                if stock_warning_threshold is not None:
                    update_fields.append("stock_warning_threshold = %s")
                    params.append(stock_warning_threshold)
                
                if update_fields:
                    update_fields.append("updated_at = CURRENT_TIMESTAMP")
                    params.append(card_id)
                    
                    sql = f"UPDATE cards SET {', '.join(update_fields)} WHERE id = %s"
                    cursor.execute(sql, params)
                    
                    if cursor.rowcount > 0:
                        self.conn.commit()
                        logger.info(f"卡券更新成功: {card_id}")
                        return True
                    else:
                        logger.warning(f"卡券不存在: {card_id}")
                        return False
                else:
                    logger.warning("没有提供要更新的字段")
                    return False
                    
            except Exception as e:
                logger.error(f"更新卡券失败: {e}")
                self.conn.rollback()
                return False

    def delete_card(self, card_id: int):
        """删除卡券"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('DELETE FROM cards WHERE id = %s', (card_id,))
                
                if cursor.rowcount > 0:
                    self.conn.commit()
                    logger.info(f"卡券删除成功: {card_id}")
                    return True
                else:
                    logger.warning(f"卡券不存在: {card_id}")
                    return False
            except Exception as e:
                logger.error(f"删除卡券失败: {e}")
                self.conn.rollback()
                return False

    def update_card_image_url(self, card_id: int, new_image_url: str) -> bool:
        """更新卡券的图片URL"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()

                # 更新图片URL
                self._execute_sql(cursor,
                    "UPDATE cards SET image_url = %s, updated_at = CURRENT_TIMESTAMP WHERE id = %s AND type = 'image'",
                    (new_image_url, card_id))

                self.conn.commit()

                # 检查是否有行被更新
                if cursor.rowcount > 0:
                    logger.info(f"卡券图片URL更新成功: 卡券ID: {card_id}, 新URL: {new_image_url}")
                    return True
                else:
                    logger.warning(f"未找到匹配的图片卡券: 卡券ID: {card_id}")
                    return False

            except Exception as e:
                logger.error(f"更新卡券图片URL失败: {e}")
                self.conn.rollback()
                return False

    def consume_batch_data(self, card_id: int):
        """消费批量数据的第一条记录（线程安全）"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()

                # 获取卡券的批量数据和库存预警阈值
                self._execute_sql(cursor, "SELECT data_content, stock_warning_threshold FROM cards WHERE id = %s AND type = 'data'", (card_id,))
                result = cursor.fetchone()

                if not result or not result[0]:
                    logger.warning(f"卡券 {card_id} 没有批量数据")
                    return None

                data_content, stock_warning_threshold = result[0], result[1] or 0
                lines = [line.strip() for line in data_content.split('\n') if line.strip()]

                if not lines:
                    logger.warning(f"卡券 {card_id} 批量数据为空")
                    return None

                # 获取第一条数据
                first_line = lines[0]

                # 移除第一条数据，更新数据库
                remaining_lines = lines[1:]
                new_data_content = '\n'.join(remaining_lines)

                self._execute_sql(cursor, '''
                UPDATE cards
                SET data_content = %s, updated_at = CURRENT_TIMESTAMP
                WHERE id = %s
                ''', (new_data_content, card_id))

                self.conn.commit()

                # 检查库存预警
                remaining_stock = len(remaining_lines)
                logger.info(f"消费批量数据成功: 卡券ID={card_id}, 剩余={remaining_stock}条")
                
                # 如果启用了库存预警且剩余库存小于等于预警阈值，触发预警
                if stock_warning_threshold > 0 and remaining_stock <= stock_warning_threshold:
                    logger.warning(f"卡券库存预警: 卡券ID={card_id}, 剩余库存={remaining_stock}, 预警阈值={stock_warning_threshold}")
                    # 这里触发库存预警通知，需要异步处理
                    self._trigger_stock_warning_async(card_id, remaining_stock, stock_warning_threshold)
                
                return first_line

            except Exception as e:
                logger.error(f"消费批量数据失败: {e}")
                self.conn.rollback()
                return None

    def get_card_stock_info(self, card_id: int):
        """获取卡券库存信息"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                self._execute_sql(cursor, """
                SELECT name, data_content, stock_warning_threshold 
                FROM cards 
                WHERE id = %s AND type = 'data'
                """, (card_id,))
                result = cursor.fetchone()
                
                if not result:
                    return None
                
                name, data_content, stock_warning_threshold = result
                
                # 计算当前库存
                if data_content:
                    lines = [line.strip() for line in data_content.split('\n') if line.strip()]
                    current_stock = len(lines)
                else:
                    current_stock = 0
                
                return {
                    'card_id': card_id,
                    'name': name,
                    'current_stock': current_stock,
                    'warning_threshold': stock_warning_threshold or 0,
                    'is_low_stock': stock_warning_threshold > 0 and current_stock <= stock_warning_threshold
                }
                
            except Exception as e:
                logger.error(f"获取卡券库存信息失败: {e}")
                return None

    def _trigger_stock_warning_async(self, card_id: int, remaining_stock: int, threshold: int):
        """异步触发库存预警通知"""
        import asyncio
        import threading
        
        def run_async_warning():
            try:
                loop = asyncio.new_event_loop()
                asyncio.set_event_loop(loop)
                loop.run_until_complete(self._send_stock_warning_notification(card_id, remaining_stock, threshold))
                loop.close()
            except Exception as e:
                logger.error(f"异步发送库存预警通知失败: {e}")
        
        # 在新线程中运行异步任务
        thread = threading.Thread(target=run_async_warning)
        thread.daemon = True
        thread.start()

    async def _send_stock_warning_notification(self, card_id: int, remaining_stock: int, threshold: int):
        """发送库存预警通知"""
        try:
            # 获取卡券信息
            stock_info = self.get_card_stock_info(card_id)
            if not stock_info:
                return
            
            card_name = stock_info['name']
            
            # 构建预警消息
            warning_message = f"🚨 库存预警通知\n\n" \
                            f"卡券名称: {card_name}\n" \
                            f"卡券ID: {card_id}\n" \
                            f"剩余库存: {remaining_stock} 条\n" \
                            f"预警阈值: {threshold} 条\n" \
                            f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n" \
                            f"请及时补充库存！"
            
            # 获取所有启用的通知渠道配置
            notification_channels = self.get_notification_channels()
            
            for channel in notification_channels:
                if not channel.get('enabled', False):
                    continue
                    
                channel_type = channel.get('type')
                config_str = channel.get('config', '{}')
                
                # 解析配置JSON
                try:
                    import json
                    config_data = json.loads(config_str) if isinstance(config_str, str) else config_str
                except (json.JSONDecodeError, TypeError):
                    logger.warning(f"解析通知渠道配置失败: {config_str}")
                    continue
                
                if channel_type == 'webhook' and config_data.get('api_format') == 'wechat_bot':
                    # 使用微信机器人发送预警
                    await self._send_wechat_stock_warning(config_data, warning_message)
                    
        except Exception as e:
            logger.error(f"发送库存预警通知失败: {e}")

    async def _send_wechat_stock_warning(self, config_data: dict, message: str):
        """通过微信机器人发送库存预警"""
        try:
            import aiohttp
            import json
            
            webhook_url = config_data.get('webhook_url', '')
            wxid_param = config_data.get('wxid_param', '')
            target_wxid = config_data.get('target_wxid', '')
            
            if not webhook_url or not target_wxid:
                logger.warning("微信机器人配置不完整，无法发送库存预警")
                return
            
            # 构建完整的URL
            final_url = webhook_url
            if wxid_param:
                separator = '&' if '?' in webhook_url else '?'
                final_url = f"{webhook_url}{separator}wxid={wxid_param}"
            
            # 构建请求数据
            data = {
                "type": "sendText",
                "data": {
                    "wxid": target_wxid,
                    "msg": message
                }
            }
            
            # 发送请求
            async with aiohttp.ClientSession() as session:
                async with session.post(final_url, json=data, timeout=10) as response:
                    if response.status == 200:
                        logger.info(f"库存预警通知发送成功 - 微信群: {target_wxid}")
                    else:
                        logger.warning(f"库存预警通知发送失败: HTTP {response.status}")
                        
        except Exception as e:
            logger.error(f"发送微信库存预警异常: {e}")

    # -------------------- Cookie状态管理 --------------------
    def save_cookie_status(self, cookie_id: str, enabled: bool):
        """保存Cookie状态"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                INSERT INTO cookie_status (cookie_id, enabled)
                VALUES (%s, %s)
                ON DUPLICATE KEY UPDATE
                enabled = VALUES(enabled),
                updated_at = CURRENT_TIMESTAMP
                ''', (cookie_id, enabled))
                self.conn.commit()
                logger.debug(f"保存Cookie状态: {cookie_id} -> {'启用' if enabled else '禁用'}")
            except Exception as e:
                logger.error(f"保存Cookie状态失败: {e}")
                raise

    def get_cookie_status(self, cookie_id: str) -> bool:
        """获取Cookie状态"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('SELECT enabled FROM cookie_status WHERE cookie_id = %s', (cookie_id,))
                result = cursor.fetchone()
                return bool(result[0]) if result else True  # 默认启用
            except Exception as e:
                logger.error(f"获取Cookie状态失败: {e}")
                return True

    def get_all_cookie_status(self) -> Dict[str, bool]:
        """获取所有Cookie状态"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('SELECT cookie_id, enabled FROM cookie_status')
                results = cursor.fetchall()
                return {row[0]: bool(row[1]) for row in results}
            except Exception as e:
                logger.error(f"获取所有Cookie状态失败: {e}")
                return {}

    # -------------------- 暂停聊天管理 --------------------
    def add_permanently_paused_chat(self, chat_id: str, cookie_id: str, reason: str = "转人工") -> bool:
        """添加永久暂停的聊天"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                INSERT IGNORE INTO permanently_paused_chats (chat_id, cookie_id, reason)
                VALUES (%s, %s, %s)
                ''', (chat_id, cookie_id, reason))
                self.conn.commit()
                logger.info(f"添加永久暂停聊天: {chat_id} (原因: {reason})")
                return True
            except Exception as e:
                logger.error(f"添加永久暂停聊天失败: {e}")
                self.conn.rollback()
                return False

    def remove_permanently_paused_chat(self, chat_id: str) -> bool:
        """移除永久暂停的聊天"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('DELETE FROM permanently_paused_chats WHERE chat_id = %s', (chat_id,))
                self.conn.commit()
                logger.info(f"移除永久暂停聊天: {chat_id}")
                return True
            except Exception as e:
                logger.error(f"移除永久暂停聊天失败: {e}")
                self.conn.rollback()
                return False

    def get_permanently_paused_chats(self) -> List[str]:
        """获取所有永久暂停的聊天ID"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('SELECT chat_id FROM permanently_paused_chats')
                results = cursor.fetchall()
                return [row[0] for row in results]
            except Exception as e:
                logger.error(f"获取永久暂停聊天失败: {e}")
                return []

    def get_permanently_paused_chats_by_cookie(self, cookie_id: str) -> List[dict]:
        """获取指定Cookie的永久暂停聊天"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT chat_id, reason, created_at FROM permanently_paused_chats 
                WHERE cookie_id = %s ORDER BY created_at DESC
                ''', (cookie_id,))
                results = cursor.fetchall()
                return [{'chat_id': row[0], 'reason': row[1], 'created_at': row[2]} for row in results]
            except Exception as e:
                logger.error(f"获取指定Cookie的永久暂停聊天失败: {e}")
                return []

    def is_chat_permanently_paused(self, chat_id: str) -> bool:
        """检查聊天是否被永久暂停"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('SELECT 1 FROM permanently_paused_chats WHERE chat_id = %s', (chat_id,))
                return cursor.fetchone() is not None
            except Exception as e:
                logger.error(f"检查聊天暂停状态失败: {e}")
                return False

    # -------------------- Token状态管理 --------------------
    def update_token_status(self, cookie_id: str, token_status: str, token_failed_at: str = None) -> bool:
        """更新Token状态"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                UPDATE cookies SET token_status = %s, token_failed_at = %s, updated_at = CURRENT_TIMESTAMP
                WHERE id = %s
                ''', (token_status, token_failed_at, cookie_id))
                
                if cursor.rowcount > 0:
                    self.conn.commit()
                    logger.info(f"Token状态更新成功: {cookie_id} -> {token_status}")
                    return True
                else:
                    logger.warning(f"Cookie不存在: {cookie_id}")
                    return False
            except Exception as e:
                logger.error(f"更新Token状态失败: {e}")
                self.conn.rollback()
                return False

    def get_token_status(self, cookie_id: str) -> str:
        """获取Token状态"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('SELECT token_status FROM cookies WHERE id = %s', (cookie_id,))
                result = cursor.fetchone()
                return result[0] if result else 'unknown'
            except Exception as e:
                logger.error(f"获取Token状态失败: {e}")
                return 'unknown'

    # -------------------- Cookie详细信息管理 --------------------
    def get_cookie_details(self, cookie_id: str) -> Optional[Dict[str, any]]:
        """获取Cookie详细信息"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT id, value, user_id, remark, auto_confirm, pause_duration, 
                       pause_enabled, token_status, token_failed_at, notification_wxid,
                       created_at, updated_at
                FROM cookies WHERE id = %s
                ''', (cookie_id,))
                result = cursor.fetchone()
                
                if result:
                    return {
                        'id': result[0],
                        'value': result[1],
                        'user_id': result[2],
                        'remark': result[3],
                        'auto_confirm': bool(result[4]),
                        'pause_duration': result[5],
                        'pause_enabled': bool(result[6]),
                        'token_status': result[7],
                        'token_failed_at': result[8],
                        'notification_wxid': result[9],
                        'created_at': result[10],
                        'updated_at': result[11]
                    }
                return None
            except Exception as e:
                logger.error(f"获取Cookie详细信息失败: {e}")
                return None

    def update_auto_confirm(self, cookie_id: str, auto_confirm: bool) -> bool:
        """更新自动确认设置"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                UPDATE cookies SET auto_confirm = %s, updated_at = CURRENT_TIMESTAMP
                WHERE id = %s
                ''', (auto_confirm, cookie_id))
                self.conn.commit()
                return True
            except Exception as e:
                logger.error(f"更新自动确认设置失败: {e}")
                self.conn.rollback()
                return False

    def update_cookie_remark(self, cookie_id: str, remark: str) -> bool:
        """更新Cookie备注"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                UPDATE cookies SET remark = %s, updated_at = CURRENT_TIMESTAMP
                WHERE id = %s
                ''', (remark, cookie_id))
                self.conn.commit()
                return True
            except Exception as e:
                logger.error(f"更新Cookie备注失败: {e}")
                self.conn.rollback()
                return False

    def update_cookie_pause_duration(self, cookie_id: str, pause_duration: int) -> bool:
        """更新暂停时长"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                UPDATE cookies SET pause_duration = %s, updated_at = CURRENT_TIMESTAMP
                WHERE id = %s
                ''', (pause_duration, cookie_id))
                self.conn.commit()
                return True
            except Exception as e:
                logger.error(f"更新暂停时长失败: {e}")
                self.conn.rollback()
                return False

    def update_cookie_notification_wxid(self, cookie_id: str, notification_wxid: str) -> bool:
        """更新通知微信ID"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                UPDATE cookies SET notification_wxid = %s, updated_at = CURRENT_TIMESTAMP
                WHERE id = %s
                ''', (notification_wxid, cookie_id))
                self.conn.commit()
                return True
            except Exception as e:
                logger.error(f"更新通知微信ID失败: {e}")
                self.conn.rollback()
                return False

    def get_cookie_pause_duration(self, cookie_id: str) -> int:
        """获取暂停时长"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('SELECT pause_duration FROM cookies WHERE id = %s', (cookie_id,))
                result = cursor.fetchone()
                return result[0] if result else 10
            except Exception as e:
                logger.error(f"获取暂停时长失败: {e}")
                return 10

    def get_auto_confirm(self, cookie_id: str) -> bool:
        """获取自动确认设置"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('SELECT auto_confirm FROM cookies WHERE id = %s', (cookie_id,))
                result = cursor.fetchone()
                return bool(result[0]) if result else False
            except Exception as e:
                logger.error(f"获取自动确认设置失败: {e}")
                return False

    def update_pause_enabled(self, cookie_id: str, pause_enabled: bool) -> bool:
        """更新暂停启用状态"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                UPDATE cookies SET pause_enabled = %s, updated_at = CURRENT_TIMESTAMP
                WHERE id = %s
                ''', (pause_enabled, cookie_id))
                self.conn.commit()
                return True
            except Exception as e:
                logger.error(f"更新暂停启用状态失败: {e}")
                self.conn.rollback()
                return False

    def get_pause_enabled(self, cookie_id: str) -> bool:
        """获取暂停启用状态"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('SELECT pause_enabled FROM cookies WHERE id = %s', (cookie_id,))
                result = cursor.fetchone()
                return bool(result[0]) if result else True
            except Exception as e:
                logger.error(f"获取暂停启用状态失败: {e}")
                return True

    def get_all_users(self):
        """获取所有用户"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT id, username, email, is_active, is_admin, created_at, updated_at
                FROM users ORDER BY created_at DESC
                ''')
                
                users = []
                for row in cursor.fetchall():
                    users.append({
                        'id': row[0],
                        'username': row[1],
                        'email': row[2],
                        'is_active': bool(row[3]),
                        'is_admin': bool(row[4]),
                        'created_at': row[5],
                        'updated_at': row[6]
                    })
                
                return users
            except Exception as e:
                logger.error(f"获取所有用户失败: {e}")
                return []

    def get_user_by_id(self, user_id: int):
        """根据用户ID获取用户信息"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT id, username, email, password_hash, phone, nickname, is_active, is_admin, created_at, updated_at
                FROM users WHERE id = %s
                ''', (user_id,))
                result = cursor.fetchone()
                
                if result:
                    return {
                        'id': result[0],
                        'username': result[1],
                        'email': result[2],
                        'password_hash': result[3],
                        'phone': result[4],
                        'nickname': result[5],
                        'is_active': bool(result[6]),
                        'is_admin': bool(result[7]),
                        'created_at': result[8],
                        'updated_at': result[9]
                    }
                return None
            except Exception as e:
                logger.error(f"根据用户ID获取用户失败: {e}")
                return None

    def delete_user_and_data(self, user_id: int):
        """删除用户及其所有相关数据"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 获取用户的所有cookies
                cursor.execute('SELECT id FROM cookies WHERE user_id = %s', (user_id,))
                cookie_ids = [row[0] for row in cursor.fetchall()]
                
                # 删除所有相关数据（外键约束会自动删除关联数据）
                for cookie_id in cookie_ids:
                    cursor.execute('DELETE FROM cookies WHERE id = %s', (cookie_id,))
                
                # 删除用户记录
                cursor.execute('DELETE FROM users WHERE id = %s', (user_id,))
                
                self.conn.commit()
                logger.info(f"用户及其数据删除成功: {user_id}")
                return True
            except Exception as e:
                logger.error(f"删除用户及其数据失败: {e}")
                self.conn.rollback()
                return False

    # -------------------- 关键词操作补充 --------------------
    def get_all_keywords(self, user_id: int = None) -> Dict[str, List[Tuple[str, str]]]:
        """获取所有关键字"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                if user_id is not None:
                    cursor.execute('''
                    SELECT c.id, k.keyword, k.reply
                    FROM cookies c
                    JOIN keywords k ON c.id = k.cookie_id
                    WHERE c.user_id = %s
                    ORDER BY c.id, k.keyword
                    ''', (user_id,))
                else:
                    cursor.execute('''
                    SELECT cookie_id, keyword, reply
                    FROM keywords
                    ORDER BY cookie_id, keyword
                    ''')
                
                result = {}
                for row in cursor.fetchall():
                    cookie_id = row[0]
                    if cookie_id not in result:
                        result[cookie_id] = []
                    result[cookie_id].append((row[1], row[2]))
                
                return result
            except Exception as e:
                logger.error(f"获取所有关键字失败: {e}")
                return {}

    def update_keyword_image_url(self, cookie_id: str, keyword: str, new_image_url: str) -> bool:
        """更新关键词的图片URL"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                UPDATE keywords SET image_url = %s
                WHERE cookie_id = %s AND keyword = %s AND type = 'image'
                ''', (new_image_url, cookie_id, keyword))
                
                if cursor.rowcount > 0:
                    self.conn.commit()
                    logger.info(f"关键词图片URL更新成功: {cookie_id} - {keyword}")
                    return True
                else:
                    logger.warning(f"未找到要更新的关键词: {cookie_id} - {keyword}")
                    return False
            except Exception as e:
                logger.error(f"更新关键词图片URL失败: {e}")
                self.conn.rollback()
                return False

    def delete_keyword_by_index(self, cookie_id: str, index: int) -> bool:
        """根据索引删除关键词"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                # 获取按顺序排列的关键词
                cursor.execute('''
                SELECT id FROM keywords 
                WHERE cookie_id = %s 
                ORDER BY id 
                LIMIT 1 OFFSET %s
                ''', (cookie_id, index))
                
                result = cursor.fetchone()
                if result:
                    keyword_id = result[0]
                    cursor.execute('DELETE FROM keywords WHERE id = %s', (keyword_id,))
                    self.conn.commit()
                    logger.info(f"关键词删除成功: {cookie_id} - 索引 {index}")
                    return True
                else:
                    logger.warning(f"关键词索引不存在: {cookie_id} - 索引 {index}")
                    return False
            except Exception as e:
                logger.error(f"删除关键词失败: {e}")
                self.conn.rollback()
                return False

    def delete_keywords_by_item_id(self, cookie_id: str, item_id: str) -> bool:
        """删除指定商品的所有关键词"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                DELETE FROM keywords 
                WHERE cookie_id = %s AND item_id = %s
                ''', (cookie_id, item_id))
                
                deleted_count = cursor.rowcount
                self.conn.commit()
                logger.info(f"商品关键词删除成功: {cookie_id} - {item_id}, 删除 {deleted_count} 条")
                return True
            except Exception as e:
                logger.error(f"删除商品关键词失败: {e}")
                self.conn.rollback()
                return False

    def has_item_keywords(self, cookie_id: str, item_id: str) -> bool:
        """检查商品是否有关键词"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT COUNT(*) FROM keywords 
                WHERE cookie_id = %s AND item_id = %s
                ''', (cookie_id, item_id))
                
                count = cursor.fetchone()[0]
                return count > 0
            except Exception as e:
                logger.error(f"检查商品关键词失败: {e}")
                return False

    # -------------------- 统计和分析方法 --------------------
    def get_account_summary_stats(self, user_id: int = None) -> Dict[str, int]:
        """获取账号汇总统计"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                if user_id is not None:
                    # 使用一个复杂查询来准确计算所有统计数据，改为统计商品数量
                    cursor.execute('''
                    SELECT 
                        COUNT(DISTINCT c.id) as total_accounts,
                        COUNT(DISTINCT CASE WHEN COALESCE(cs.enabled, TRUE) = TRUE THEN c.id END) as enabled_accounts,
                        COUNT(DISTINCT CASE WHEN COALESCE(cs.enabled, TRUE) = TRUE AND k.item_id IS NOT NULL AND k.item_id != '' THEN c.id END) as active_accounts,
                        COUNT(DISTINCT CASE WHEN k.item_id IS NOT NULL AND k.item_id != '' THEN k.item_id END) as total_items,
                        COUNT(DISTINCT CASE WHEN k.item_id IS NOT NULL AND k.item_id != '' AND (k.type = 'text' OR k.type IS NULL) THEN k.item_id END) as text_items,
                        COUNT(DISTINCT CASE WHEN k.item_id IS NOT NULL AND k.item_id != '' AND k.type = 'image' THEN k.item_id END) as image_items
                    FROM cookies c
                    LEFT JOIN cookie_status cs ON c.id = cs.cookie_id
                    LEFT JOIN keywords k ON c.id = k.cookie_id
                    WHERE c.user_id = %s
                    ''', (user_id,))
                else:
                    cursor.execute('''
                    SELECT 
                        COUNT(DISTINCT c.id) as total_accounts,
                        COUNT(DISTINCT CASE WHEN COALESCE(cs.enabled, TRUE) = TRUE THEN c.id END) as enabled_accounts,
                        COUNT(DISTINCT CASE WHEN COALESCE(cs.enabled, TRUE) = TRUE AND k.item_id IS NOT NULL AND k.item_id != '' THEN c.id END) as active_accounts,
                        COUNT(DISTINCT CASE WHEN k.item_id IS NOT NULL AND k.item_id != '' THEN k.item_id END) as total_items,
                        COUNT(DISTINCT CASE WHEN k.item_id IS NOT NULL AND k.item_id != '' AND (k.type = 'text' OR k.type IS NULL) THEN k.item_id END) as text_items,
                        COUNT(DISTINCT CASE WHEN k.item_id IS NOT NULL AND k.item_id != '' AND k.type = 'image' THEN k.item_id END) as image_items
                    FROM cookies c
                    LEFT JOIN cookie_status cs ON c.id = cs.cookie_id
                    LEFT JOIN keywords k ON c.id = k.cookie_id
                    ''')
                
                row = cursor.fetchone()
                if row:
                    return {
                        'total_accounts': row[0],
                        'enabled_accounts': row[1],
                        'active_accounts': row[2],
                        'total_keywords': row[3],  # 保持字段名不变，但实际是商品数量
                        'text_keywords': row[4],   # 保持字段名不变，但实际是文本商品数量
                        'image_keywords': row[5]   # 保持字段名不变，但实际是图片商品数量
                    }
                else:
                    return {
                        'total_accounts': 0,
                        'enabled_accounts': 0,
                        'active_accounts': 0,
                        'total_keywords': 0,
                        'text_keywords': 0,
                        'image_keywords': 0
                    }
                
            except Exception as e:
                logger.error(f"获取账号汇总统计失败: {e}")
                return {
                    'total_accounts': 0,
                    'total_keywords': 0,
                    'total_items': 0,
                    'total_orders': 0
                }

    def get_keywords_with_pagination(self, cookie_id: str, page: int = 1, page_size: int = 50) -> Dict[str, Any]:
        """分页获取关键词"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 获取总数
                cursor.execute('SELECT COUNT(*) FROM keywords WHERE cookie_id = %s', (cookie_id,))
                total_count = cursor.fetchone()[0]
                
                # 计算分页
                total_pages = (total_count + page_size - 1) // page_size
                offset = (page - 1) * page_size
                
                # 获取分页数据（包含词库信息）
                cursor.execute('''
                SELECT k.keyword, k.reply, k.item_id, k.type, k.image_url, k.created_at, k.library_id, kl.name as library_name
                FROM keywords k 
                LEFT JOIN keyword_libraries kl ON k.library_id = kl.id 
                WHERE k.cookie_id = %s
                ORDER BY k.created_at DESC
                LIMIT %s OFFSET %s
                ''', (cookie_id, page_size, offset))
                
                keywords = []
                for row in cursor.fetchall():
                    keywords.append({
                        'keyword': row[0],
                        'reply': row[1],
                        'item_id': row[2],
                        'type': row[3] or 'text',
                        'image_url': row[4],
                        'created_at': row[5],
                        'library_id': row[6],
                        'library_name': row[7]
                    })
                
                return {
                    'keywords': keywords,
                    'total_count': total_count,
                    'total_pages': total_pages,
                    'current_page': page,
                    'page_size': page_size
                }
                
            except Exception as e:
                logger.error(f"分页获取关键词失败: {e}")
                return {
                    'keywords': [],
                    'total_count': 0,
                    'total_pages': 0,
                    'current_page': page,
                    'page_size': page_size
                }

    # -------------------- 延迟引导任务管理 --------------------
    def create_delayed_guide_task(self, cookie_id: str, user_id: str, chat_id: str, 
                                 item_id: str = None, trigger_time: str = None, 
                                 scheduled_time: str = None) -> bool:
        """创建延迟引导任务"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                INSERT INTO delayed_guide_tasks 
                (cookie_id, user_id, chat_id, item_id, trigger_time, scheduled_time, status)
                VALUES (%s, %s, %s, %s, %s, %s, 'pending')
                ''', (cookie_id, user_id, chat_id, item_id, trigger_time, scheduled_time))
                
                self.conn.commit()
                task_id = cursor.lastrowid
                logger.info(f"延迟引导任务创建成功: {task_id}")
                return True
            except Exception as e:
                logger.error(f"创建延迟引导任务失败: {e}")
                self.conn.rollback()
                return False

    def get_pending_delayed_guide_tasks(self) -> List[Dict[str, Any]]:
        """获取待处理的延迟引导任务"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT id, cookie_id, user_id, chat_id, item_id, trigger_time, 
                       scheduled_time, status, created_at
                FROM delayed_guide_tasks 
                WHERE status = 'pending' AND scheduled_time <= NOW()
                ORDER BY scheduled_time ASC
                ''')
                
                tasks = []
                for row in cursor.fetchall():
                    tasks.append({
                        'id': row[0],
                        'cookie_id': row[1],
                        'user_id': row[2],
                        'chat_id': row[3],
                        'item_id': row[4],
                        'trigger_time': row[5],
                        'scheduled_time': row[6],
                        'status': row[7],
                        'created_at': row[8]
                    })
                
                return tasks
            except Exception as e:
                logger.error(f"获取待处理延迟引导任务失败: {e}")
                return []

    def update_delayed_guide_task_status(self, task_id: int, status: str, 
                                       sent_at: str = None, cancelled_at: str = None) -> bool:
        """更新延迟引导任务状态"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                if status == 'sent' and sent_at:
                    cursor.execute('''
                    UPDATE delayed_guide_tasks 
                    SET status = %s, sent_at = %s 
                    WHERE id = %s
                    ''', (status, sent_at, task_id))
                elif status == 'cancelled' and cancelled_at:
                    cursor.execute('''
                    UPDATE delayed_guide_tasks 
                    SET status = %s, cancelled_at = %s 
                    WHERE id = %s
                    ''', (status, cancelled_at, task_id))
                else:
                    cursor.execute('''
                    UPDATE delayed_guide_tasks 
                    SET status = %s 
                    WHERE id = %s
                    ''', (status, task_id))
                
                if cursor.rowcount > 0:
                    self.conn.commit()
                    logger.info(f"延迟引导任务状态更新成功: {task_id} -> {status}")
                    return True
                else:
                    logger.warning(f"延迟引导任务不存在: {task_id}")
                    return False
            except Exception as e:
                logger.error(f"更新延迟引导任务状态失败: {e}")
                self.conn.rollback()
                return False

    def get_delayed_guide_tasks_by_cookie(self, cookie_id: str, status: str = None) -> List[Dict[str, Any]]:
        """获取指定Cookie的延迟引导任务"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                if status:
                    cursor.execute('''
                    SELECT id, cookie_id, user_id, chat_id, item_id, trigger_time,
                           scheduled_time, status, sent_at, cancelled_at, created_at
                    FROM delayed_guide_tasks 
                    WHERE cookie_id = %s AND status = %s
                    ORDER BY created_at DESC
                    ''', (cookie_id, status))
                else:
                    cursor.execute('''
                    SELECT id, cookie_id, user_id, chat_id, item_id, trigger_time,
                           scheduled_time, status, sent_at, cancelled_at, created_at
                    FROM delayed_guide_tasks 
                    WHERE cookie_id = %s
                    ORDER BY created_at DESC
                    ''', (cookie_id,))
                
                tasks = []
                for row in cursor.fetchall():
                    tasks.append({
                        'id': row[0],
                        'cookie_id': row[1],
                        'user_id': row[2],
                        'chat_id': row[3],
                        'item_id': row[4],
                        'trigger_time': row[5],
                        'scheduled_time': row[6],
                        'status': row[7],
                        'sent_at': row[8],
                        'cancelled_at': row[9],
                        'created_at': row[10]
                    })
                
                return tasks
            except Exception as e:
                logger.error(f"获取Cookie延迟引导任务失败: {e}")
                return []

    def cleanup_old_delayed_guide_tasks(self, days_old: int = 30) -> bool:
        """清理旧的延迟引导任务记录"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                DELETE FROM delayed_guide_tasks 
                WHERE created_at < DATE_SUB(NOW(), INTERVAL %s DAY)
                AND status IN ('sent', 'cancelled', 'failed')
                ''', (days_old,))
                
                deleted_count = cursor.rowcount
                self.conn.commit()
                logger.info(f"清理旧延迟引导任务记录: 删除 {deleted_count} 条")
                return True
            except Exception as e:
                logger.error(f"清理旧延迟引导任务记录失败: {e}")
                self.conn.rollback()
                return False

    def cancel_delayed_guide_tasks_by_chat(self, chat_id: str, cookie_id: str = None) -> bool:
        """取消指定聊天的所有待处理延迟引导任务"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                if cookie_id:
                    cursor.execute('''
                    UPDATE delayed_guide_tasks 
                    SET status = 'cancelled', cancelled_at = NOW()
                    WHERE chat_id = %s AND cookie_id = %s AND status = 'pending'
                    ''', (chat_id, cookie_id))
                else:
                    cursor.execute('''
                    UPDATE delayed_guide_tasks 
                    SET status = 'cancelled', cancelled_at = NOW()
                    WHERE chat_id = %s AND status = 'pending'
                    ''', (chat_id,))
                
                cancelled_count = cursor.rowcount
                self.conn.commit()
                logger.info(f"取消聊天延迟引导任务: {chat_id}, 取消 {cancelled_count} 个任务")
                return True
            except Exception as e:
                logger.error(f"取消聊天延迟引导任务失败: {e}")
                self.conn.rollback()
                return False

    # -------------------- 未匹配回复计数管理 --------------------
    def increment_unmatch_reply_count(self, cookie_id: str, user_id: str, chat_id: str) -> int:
        """增加未匹配回复计数"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                INSERT INTO unmatch_reply_counts (cookie_id, user_id, chat_id, count)
                VALUES (%s, %s, %s, 1)
                ON DUPLICATE KEY UPDATE
                count = count + 1,
                updated_at = CURRENT_TIMESTAMP
                ''', (cookie_id, user_id, chat_id))
                
                # 获取当前计数
                cursor.execute('''
                SELECT count FROM unmatch_reply_counts 
                WHERE cookie_id = %s AND user_id = %s AND chat_id = %s
                ''', (cookie_id, user_id, chat_id))
                
                result = cursor.fetchone()
                count = result[0] if result else 0
                
                self.conn.commit()
                logger.debug(f"未匹配回复计数增加: {cookie_id}-{chat_id} -> {count}")
                return count
            except Exception as e:
                logger.error(f"增加未匹配回复计数失败: {e}")
                self.conn.rollback()
                return 0

    def get_unmatch_reply_count(self, cookie_id: str, user_id: str, chat_id: str) -> int:
        """获取未匹配回复计数"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT count FROM unmatch_reply_counts 
                WHERE cookie_id = %s AND user_id = %s AND chat_id = %s
                ''', (cookie_id, user_id, chat_id))
                
                result = cursor.fetchone()
                return result[0] if result else 0
            except Exception as e:
                logger.error(f"获取未匹配回复计数失败: {e}")
                return 0

    def reset_unmatch_reply_count(self, cookie_id: str, user_id: str, chat_id: str) -> bool:
        """重置未匹配回复计数"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                DELETE FROM unmatch_reply_counts 
                WHERE cookie_id = %s AND user_id = %s AND chat_id = %s
                ''', (cookie_id, user_id, chat_id))
                
                self.conn.commit()
                logger.debug(f"未匹配回复计数重置: {cookie_id}-{chat_id}")
                return True
            except Exception as e:
                logger.error(f"重置未匹配回复计数失败: {e}")
                self.conn.rollback()
                return False

    def check_unmatch_reply_limit(self, cookie_id: str, user_id: str, chat_id: str, max_count: int = 3) -> bool:
        """检查是否可以发送未匹配回复（是否超过次数限制）"""
        current_count = self.get_unmatch_reply_count(cookie_id, user_id, chat_id)
        return current_count < max_count

    def has_sent_final_reminder(self, cookie_id: str, user_id: str, chat_id: str) -> bool:
        """检查是否已发送最终提醒消息"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT final_reminder_sent FROM unmatch_reply_counts 
                WHERE cookie_id = %s AND user_id = %s AND chat_id = %s
                ''', (cookie_id, user_id, chat_id))
                
                result = cursor.fetchone()
                return result[0] if result else False
            except Exception as e:
                logger.error(f"检查最终提醒消息状态失败: {e}")
                return False

    def mark_final_reminder_sent(self, cookie_id: str, user_id: str, chat_id: str) -> bool:
        """标记已发送最终提醒消息"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                UPDATE unmatch_reply_counts 
                SET final_reminder_sent = TRUE, updated_at = CURRENT_TIMESTAMP
                WHERE cookie_id = %s AND user_id = %s AND chat_id = %s
                ''', (cookie_id, user_id, chat_id))
                
                self.conn.commit()
                logger.debug(f"标记最终提醒消息已发送: {cookie_id}-{chat_id}")
                return True
            except Exception as e:
                logger.error(f"标记最终提醒消息失败: {e}")
                self.conn.rollback()
                return False

    # -------------------- 聊天首次回复管理 --------------------
    def mark_chat_first_reply(self, cookie_id: str, chat_id: str) -> bool:
        """标记聊天已首次回复"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                INSERT INTO chat_first_replies (cookie_id, chat_id, has_replied)
                VALUES (%s, %s, TRUE)
                ON DUPLICATE KEY UPDATE
                has_replied = TRUE,
                updated_at = CURRENT_TIMESTAMP
                ''', (cookie_id, chat_id))
                
                self.conn.commit()
                logger.debug(f"标记聊天首次回复: {cookie_id}-{chat_id}")
                return True
            except Exception as e:
                logger.error(f"标记聊天首次回复失败: {e}")
                self.conn.rollback()
                return False

    def has_chat_first_replied(self, cookie_id: str, chat_id: str) -> bool:
        """检查聊天是否已首次回复"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT has_replied FROM chat_first_replies 
                WHERE cookie_id = %s AND chat_id = %s
                ''', (cookie_id, chat_id))
                
                result = cursor.fetchone()
                return bool(result[0]) if result else False
            except Exception as e:
                logger.error(f"检查聊天首次回复状态失败: {e}")
                return False

    def clear_chat_first_reply_records(self, cookie_id: str) -> bool:
        """清空指定账号的聊天首次回复记录"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                DELETE FROM chat_first_replies WHERE cookie_id = %s
                ''', (cookie_id,))
                
                deleted_count = cursor.rowcount
                self.conn.commit()
                logger.info(f"清空聊天首次回复记录: {cookie_id}, 删除 {deleted_count} 条")
                return True
            except Exception as e:
                logger.error(f"清空聊天首次回复记录失败: {e}")
                self.conn.rollback()
                return False

    def mark_chat_has_replied(self, cookie_id: str, chat_id: str) -> bool:
        """标记对话已经回复过"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 使用 INSERT ... ON DUPLICATE KEY UPDATE 来处理插入或更新
                cursor.execute('''
                INSERT INTO chat_first_replies (cookie_id, chat_id, has_replied)
                VALUES (%s, %s, TRUE)
                ON DUPLICATE KEY UPDATE
                has_replied = TRUE,
                updated_at = CURRENT_TIMESTAMP
                ''', (cookie_id, chat_id))
                
                self.conn.commit()
                logger.info(f"标记对话已回复: 账号={cookie_id}, 对话={chat_id}")
                return True
                
            except Exception as e:
                logger.error(f"标记对话已回复失败: {e}")
                self.conn.rollback()
                return False

    # -------------------- 商品专属回复管理 --------------------
    def save_item_exclusive_reply(self, cookie_id: str, item_id: str, reply_content: str, enabled: bool = True) -> bool:
        """保存商品专属回复"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                INSERT INTO item_exclusive_replies (cookie_id, item_id, reply_content, enabled)
                VALUES (%s, %s, %s, %s)
                ON DUPLICATE KEY UPDATE
                reply_content = VALUES(reply_content),
                enabled = VALUES(enabled),
                updated_at = CURRENT_TIMESTAMP
                ''', (cookie_id, item_id, reply_content, enabled))
                
                self.conn.commit()
                logger.info(f"商品专属回复保存成功: {cookie_id}-{item_id}")
                return True
            except Exception as e:
                logger.error(f"保存商品专属回复失败: {e}")
                self.conn.rollback()
                return False

    def get_item_exclusive_reply(self, cookie_id: str, item_id: str) -> Optional[Dict[str, Any]]:
        """获取商品专属回复"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT reply_content, enabled, created_at, updated_at
                FROM item_exclusive_replies 
                WHERE cookie_id = %s AND item_id = %s
                ''', (cookie_id, item_id))
                
                result = cursor.fetchone()
                if result:
                    return {
                        'reply_content': result[0],
                        'enabled': bool(result[1]),
                        'created_at': result[2],
                        'updated_at': result[3]
                    }
                return None
            except Exception as e:
                logger.error(f"获取商品专属回复失败: {e}")
                return None

    def delete_item_exclusive_reply(self, cookie_id: str, item_id: str) -> bool:
        """删除商品专属回复"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                DELETE FROM item_exclusive_replies 
                WHERE cookie_id = %s AND item_id = %s
                ''', (cookie_id, item_id))
                
                if cursor.rowcount > 0:
                    self.conn.commit()
                    logger.info(f"商品专属回复删除成功: {cookie_id}-{item_id}")
                    return True
                else:
                    logger.warning(f"商品专属回复不存在: {cookie_id}-{item_id}")
                    return False
            except Exception as e:
                logger.error(f"删除商品专属回复失败: {e}")
                self.conn.rollback()
                return False

    def get_all_item_exclusive_replies(self, cookie_id: str) -> List[Dict[str, Any]]:
        """获取指定账号的所有商品专属回复"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT item_id, reply_content, enabled, created_at, updated_at
                FROM item_exclusive_replies 
                WHERE cookie_id = %s
                ORDER BY updated_at DESC
                ''', (cookie_id,))
                
                replies = []
                for row in cursor.fetchall():
                    replies.append({
                        'item_id': row[0],
                        'reply_content': row[1],
                        'enabled': bool(row[2]),
                        'created_at': row[3],
                        'updated_at': row[4]
                    })
                
                return replies
            except Exception as e:
                logger.error(f"获取所有商品专属回复失败: {e}")
                return []

    # -------------------- AI对话历史管理 --------------------
    def save_ai_conversation(self, cookie_id: str, chat_id: str, user_id: str, 
                           item_id: str = None, role: str = 'user', 
                           content: str = '', intent: str = None) -> bool:
        """保存AI对话历史"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                INSERT INTO ai_conversations (cookie_id, chat_id, user_id, item_id, role, content, intent)
                VALUES (%s, %s, %s, %s, %s, %s, %s)
                ''', (cookie_id, chat_id, user_id, item_id, role, content, intent))
                
                self.conn.commit()
                logger.debug(f"AI对话历史保存成功: {cookie_id}-{chat_id}")
                return True
            except Exception as e:
                logger.error(f"保存AI对话历史失败: {e}")
                self.conn.rollback()
                return False

    def get_ai_conversation_history(self, cookie_id: str, chat_id: str, limit: int = 10) -> List[Dict[str, Any]]:
        """获取AI对话历史"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT role, content, intent, created_at
                FROM ai_conversations 
                WHERE cookie_id = %s AND chat_id = %s
                ORDER BY created_at DESC
                LIMIT %s
                ''', (cookie_id, chat_id, limit))
                
                history = []
                for row in cursor.fetchall():
                    history.append({
                        'role': row[0],
                        'content': row[1],
                        'intent': row[2],
                        'created_at': row[3]
                    })
                
                return list(reversed(history))  # 按时间正序返回
            except Exception as e:
                logger.error(f"获取AI对话历史失败: {e}")
                return []

    def clear_ai_conversation_history(self, cookie_id: str, chat_id: str = None) -> bool:
        """清空AI对话历史"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                if chat_id:
                    cursor.execute('''
                    DELETE FROM ai_conversations 
                    WHERE cookie_id = %s AND chat_id = %s
                    ''', (cookie_id, chat_id))
                else:
                    cursor.execute('''
                    DELETE FROM ai_conversations 
                    WHERE cookie_id = %s
                    ''', (cookie_id,))
                
                deleted_count = cursor.rowcount
                self.conn.commit()
                logger.info(f"清空AI对话历史: {cookie_id}, 删除 {deleted_count} 条")
                return True
            except Exception as e:
                logger.error(f"清空AI对话历史失败: {e}")
                self.conn.rollback()
                return False

    # -------------------- AI商品缓存管理 --------------------
    def save_ai_item_cache(self, cookie_id: str, item_id: str, item_info: str) -> bool:
        """保存AI商品缓存"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                INSERT INTO ai_item_cache (cookie_id, item_id, item_info)
                VALUES (%s, %s, %s)
                ON DUPLICATE KEY UPDATE
                item_info = VALUES(item_info),
                updated_at = CURRENT_TIMESTAMP
                ''', (cookie_id, item_id, item_info))
                
                self.conn.commit()
                logger.debug(f"AI商品缓存保存成功: {cookie_id}-{item_id}")
                return True
            except Exception as e:
                logger.error(f"保存AI商品缓存失败: {e}")
                self.conn.rollback()
                return False

    def get_ai_item_cache(self, cookie_id: str, item_id: str) -> Optional[str]:
        """获取AI商品缓存"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT item_info FROM ai_item_cache 
                WHERE cookie_id = %s AND item_id = %s
                ''', (cookie_id, item_id))
                
                result = cursor.fetchone()
                return result[0] if result else None
            except Exception as e:
                logger.error(f"获取AI商品缓存失败: {e}")
                return None

    def clear_ai_item_cache(self, cookie_id: str, item_id: str = None) -> bool:
        """清空AI商品缓存"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                if item_id:
                    cursor.execute('''
                    DELETE FROM ai_item_cache 
                    WHERE cookie_id = %s AND item_id = %s
                    ''', (cookie_id, item_id))
                else:
                    cursor.execute('''
                    DELETE FROM ai_item_cache 
                    WHERE cookie_id = %s
                    ''', (cookie_id,))
                
                deleted_count = cursor.rowcount
                self.conn.commit()
                logger.info(f"清空AI商品缓存: {cookie_id}, 删除 {deleted_count} 条")
                return True
            except Exception as e:
                logger.error(f"清空AI商品缓存失败: {e}")
                self.conn.rollback()
                return False

    # -------------------- 批量查询优化方法 --------------------
    def get_cookies_with_keywords_batch(self, user_id: int = None) -> List[Dict[str, Any]]:
        """批量获取Cookie及其关键词信息"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 获取Cookie基本信息和统计
                if user_id is not None:
                    cursor.execute('''
                    SELECT 
                        c.id, c.value, c.user_id, c.remark, c.auto_confirm, 
                        c.pause_duration, c.pause_enabled, c.token_status, 
                        c.token_failed_at, c.notification_wxid, c.created_at, c.updated_at,
                        COALESCE(k.keyword_count, 0) as keywords_count,
                        COALESCE(i.items_count, 0) as items_count,
                        COALESCE(cs.enabled, TRUE) as status_enabled
                    FROM cookies c
                    LEFT JOIN (
                        SELECT cookie_id, COUNT(*) as keyword_count 
                        FROM keywords 
                        GROUP BY cookie_id
                    ) k ON c.id = k.cookie_id
                    LEFT JOIN (
                        SELECT cookie_id, COUNT(DISTINCT item_id) as items_count 
                        FROM keywords 
                        WHERE item_id IS NOT NULL AND item_id != ''
                        GROUP BY cookie_id
                    ) i ON c.id = i.cookie_id
                    LEFT JOIN cookie_status cs ON c.id = cs.cookie_id
                    WHERE c.user_id = %s
                    ORDER BY c.created_at DESC
                    ''', (user_id,))
                else:
                    cursor.execute('''
                    SELECT 
                        c.id, c.value, c.user_id, c.remark, c.auto_confirm, 
                        c.pause_duration, c.pause_enabled, c.token_status, 
                        c.token_failed_at, c.notification_wxid, c.created_at, c.updated_at,
                        COALESCE(k.keyword_count, 0) as keywords_count,
                        COALESCE(i.items_count, 0) as items_count,
                        COALESCE(cs.enabled, TRUE) as status_enabled
                    FROM cookies c
                    LEFT JOIN (
                        SELECT cookie_id, COUNT(*) as keyword_count 
                        FROM keywords 
                        GROUP BY cookie_id
                    ) k ON c.id = k.cookie_id
                    LEFT JOIN (
                        SELECT cookie_id, COUNT(DISTINCT item_id) as items_count 
                        FROM keywords 
                        WHERE item_id IS NOT NULL AND item_id != ''
                        GROUP BY cookie_id
                    ) i ON c.id = i.cookie_id
                    LEFT JOIN cookie_status cs ON c.id = cs.cookie_id
                    ORDER BY c.created_at DESC
                    ''')
                
                results = []
                for row in cursor.fetchall():
                    results.append({
                        'id': row[0],
                        'value': row[1],
                        'user_id': row[2],
                        'remark': row[3],
                        'auto_confirm': bool(row[4]),
                        'pause_duration': row[5],
                        'pause_enabled': bool(row[6]),
                        'token_status': row[7],
                        'token_failed_at': row[8],
                        'notification_wxid': row[9],
                        'created_at': row[10],
                        'updated_at': row[11],
                        'keywords_count': row[12],
                        'items_count': row[13],
                        'enabled': bool(row[14])
                    })
                
                return results
                
            except Exception as e:
                logger.error(f"批量获取Cookie及关键词信息失败: {e}")
                return []

    def get_cookies_with_stats_batch(self, user_id: int = None) -> List[Dict[str, Any]]:
        """批量获取Cookie及统计信息"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 获取Cookie基本信息和统计
                if user_id is not None:
                    cursor.execute('''
                    SELECT 
                        c.id, c.value, c.user_id, c.remark, c.auto_confirm, 
                        c.pause_duration, c.pause_enabled, c.token_status, 
                        c.token_failed_at, c.notification_wxid, c.created_at, c.updated_at,
                        COALESCE(k.keyword_count, 0) as keywords_count,
                        COALESCE(i.items_count, 0) as items_count,
                        COALESCE(o.orders_count, 0) as orders_count,
                        COALESCE(cs.enabled, TRUE) as status_enabled
                    FROM cookies c
                    LEFT JOIN (
                        SELECT cookie_id, COUNT(*) as keyword_count 
                        FROM keywords 
                        GROUP BY cookie_id
                    ) k ON c.id = k.cookie_id
                    LEFT JOIN (
                        SELECT cookie_id, COUNT(*) as items_count 
                        FROM item_info 
                        GROUP BY cookie_id
                    ) i ON c.id = i.cookie_id
                    LEFT JOIN (
                        SELECT cookie_id, COUNT(*) as orders_count 
                        FROM orders 
                        GROUP BY cookie_id
                    ) o ON c.id = o.cookie_id
                    LEFT JOIN cookie_status cs ON c.id = cs.cookie_id
                    WHERE c.user_id = %s
                    ORDER BY c.created_at DESC
                    ''', (user_id,))
                else:
                    cursor.execute('''
                    SELECT 
                        c.id, c.value, c.user_id, c.remark, c.auto_confirm, 
                        c.pause_duration, c.pause_enabled, c.token_status, 
                        c.token_failed_at, c.notification_wxid, c.created_at, c.updated_at,
                        COALESCE(k.keyword_count, 0) as keywords_count,
                        COALESCE(i.items_count, 0) as items_count,
                        COALESCE(o.orders_count, 0) as orders_count,
                        COALESCE(cs.enabled, TRUE) as status_enabled
                    FROM cookies c
                    LEFT JOIN (
                        SELECT cookie_id, COUNT(*) as keyword_count 
                        FROM keywords 
                        GROUP BY cookie_id
                    ) k ON c.id = k.cookie_id
                    LEFT JOIN (
                        SELECT cookie_id, COUNT(*) as items_count 
                        FROM item_info 
                        GROUP BY cookie_id
                    ) i ON c.id = i.cookie_id
                    LEFT JOIN (
                        SELECT cookie_id, COUNT(*) as orders_count 
                        FROM orders 
                        GROUP BY cookie_id
                    ) o ON c.id = o.cookie_id
                    LEFT JOIN cookie_status cs ON c.id = cs.cookie_id
                    ORDER BY c.created_at DESC
                    ''')
                
                results = []
                for row in cursor.fetchall():
                    results.append({
                        'id': row[0],
                        'value': row[1],
                        'user_id': row[2],
                        'remark': row[3],
                        'auto_confirm': bool(row[4]),
                        'pause_duration': row[5],
                        'pause_enabled': bool(row[6]),
                        'token_status': row[7],
                        'token_failed_at': row[8],
                        'notification_wxid': row[9],
                        'created_at': row[10],
                        'updated_at': row[11],
                        'keywords_count': row[12],
                        'items_count': row[13],
                        'orders_count': row[14],
                        'enabled': bool(row[15])
                    })
                
                return results
                
            except Exception as e:
                logger.error(f"批量获取Cookie及统计信息失败: {e}")
                return []

    # -------------------- 消息通知管理 --------------------
    def set_message_notification(self, cookie_id: str, channel_id: int, enabled: bool = True) -> bool:
        """设置消息通知"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                INSERT INTO message_notifications (cookie_id, channel_id, enabled)
                VALUES (%s, %s, %s)
                ON DUPLICATE KEY UPDATE
                enabled = VALUES(enabled)
                ''', (cookie_id, channel_id, enabled))
                
                self.conn.commit()
                logger.info(f"消息通知设置成功: {cookie_id} -> 渠道{channel_id} ({'启用' if enabled else '禁用'})")
                return True
            except Exception as e:
                logger.error(f"设置消息通知失败: {e}")
                self.conn.rollback()
                return False

    def get_message_notifications(self, cookie_id: str) -> List[Dict[str, Any]]:
        """获取指定账号的消息通知设置"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT mn.channel_id, mn.enabled, nc.name, nc.channel_type, nc.config
                FROM message_notifications mn
                JOIN notification_channels nc ON mn.channel_id = nc.id
                WHERE mn.cookie_id = %s
                ORDER BY nc.name
                ''', (cookie_id,))
                
                notifications = []
                for row in cursor.fetchall():
                    notifications.append({
                        'channel_id': row[0],
                        'enabled': bool(row[1]),
                        'channel_name': row[2],
                        'channel_type': row[3],
                        'config': row[4]
                    })
                
                return notifications
            except Exception as e:
                logger.error(f"获取消息通知设置失败: {e}")
                return []

    def get_all_message_notifications(self, user_id: int = None) -> Dict[str, List[Dict[str, Any]]]:
        """获取所有账号的消息通知设置"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                if user_id is not None:
                    cursor.execute('''
                    SELECT mn.cookie_id, mn.channel_id, mn.enabled, nc.name, nc.channel_type
                    FROM message_notifications mn
                    JOIN notification_channels nc ON mn.channel_id = nc.id
                    JOIN cookies c ON mn.cookie_id = c.id
                    WHERE c.user_id = %s
                    ORDER BY mn.cookie_id, nc.name
                    ''', (user_id,))
                else:
                    cursor.execute('''
                    SELECT mn.cookie_id, mn.channel_id, mn.enabled, nc.name, nc.channel_type
                    FROM message_notifications mn
                    JOIN notification_channels nc ON mn.channel_id = nc.id
                    ORDER BY mn.cookie_id, nc.name
                    ''')
                
                result = {}
                for row in cursor.fetchall():
                    cookie_id = row[0]
                    if cookie_id not in result:
                        result[cookie_id] = []
                    
                    result[cookie_id].append({
                        'channel_id': row[1],
                        'enabled': bool(row[2]),
                        'channel_name': row[3],
                        'channel_type': row[4]
                    })
                
                return result
            except Exception as e:
                logger.error(f"获取所有消息通知设置失败: {e}")
                return {}

    def delete_message_notification(self, cookie_id: str, channel_id: int) -> bool:
        """删除消息通知设置"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                DELETE FROM message_notifications 
                WHERE cookie_id = %s AND channel_id = %s
                ''', (cookie_id, channel_id))
                
                if cursor.rowcount > 0:
                    self.conn.commit()
                    logger.info(f"消息通知删除成功: {cookie_id} -> 渠道{channel_id}")
                    return True
                else:
                    logger.warning(f"消息通知不存在: {cookie_id} -> 渠道{channel_id}")
                    return False
            except Exception as e:
                logger.error(f"删除消息通知失败: {e}")
                self.conn.rollback()
                return False

    def clear_message_notifications(self, cookie_id: str) -> bool:
        """清空指定账号的所有消息通知设置"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                DELETE FROM message_notifications WHERE cookie_id = %s
                ''', (cookie_id,))
                
                deleted_count = cursor.rowcount
                self.conn.commit()
                logger.info(f"清空消息通知设置: {cookie_id}, 删除 {deleted_count} 条")
                return True
            except Exception as e:
                logger.error(f"清空消息通知设置失败: {e}")
                self.conn.rollback()
                return False

    def get_notification_enabled_channels(self, cookie_id: str) -> List[Dict[str, Any]]:
        """获取指定账号启用的通知渠道"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT nc.id, nc.name, nc.channel_type, nc.config
                FROM message_notifications mn
                JOIN notification_channels nc ON mn.channel_id = nc.id
                WHERE mn.cookie_id = %s AND mn.enabled = TRUE AND nc.enabled = TRUE
                ORDER BY nc.name
                ''', (cookie_id,))
                
                channels = []
                for row in cursor.fetchall():
                    channels.append({
                        'id': row[0],
                        'name': row[1],
                        'type': row[2],
                        'config': row[3]
                    })
                
                return channels
            except Exception as e:
                logger.error(f"获取启用的通知渠道失败: {e}")
                return []

    # -------------------- 发货规则管理 --------------------
    def create_delivery_rule(self, item_id: str, card_id: int, delivery_count: int = 1,
                           description: str = None, user_id: int = None, enabled: bool = True, 
                           generate_image: bool = True, is_multi_spec: bool = False,
                           spec_name: str = None, spec_value: str = None) -> int:
        """创建发货规则"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 如果没有指定用户ID，使用默认的admin用户
                if user_id is None:
                    cursor.execute("SELECT id FROM users WHERE username = 'admin' LIMIT 1")
                    admin_user = cursor.fetchone()
                    user_id = admin_user[0] if admin_user else 1
                
                cursor.execute('''
                INSERT INTO delivery_rules (item_id, card_id, delivery_count, description, user_id, enabled, generate_image, is_multi_spec, spec_name, spec_value)
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
                ''', (item_id, card_id, delivery_count, description, user_id, enabled, generate_image, is_multi_spec, spec_name, spec_value))
                
                rule_id = cursor.lastrowid
                self.conn.commit()
                logger.info(f"发货规则创建成功: 商品{item_id} -> 卡券{card_id}, 生成图片: {generate_image} (ID: {rule_id})")
                return rule_id
            except Exception as e:
                logger.error(f"创建发货规则失败: {e}")
                self.conn.rollback()
                raise

    def get_delivery_rules(self, user_id: int = None) -> List[Dict[str, Any]]:
        """获取所有发货规则"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                if user_id is not None:
                    cursor.execute('''
                    SELECT dr.id, dr.item_id, dr.card_id, dr.delivery_count, dr.delivery_times,
                           dr.enabled, dr.description, dr.user_id, dr.created_at, dr.updated_at,
                           c.name as card_name, c.type as card_type, dr.is_multi_spec, dr.spec_name, dr.spec_value
                    FROM delivery_rules dr
                    LEFT JOIN cards c ON dr.card_id = c.id
                    WHERE dr.user_id = %s
                    ORDER BY dr.created_at DESC
                    ''', (user_id,))
                else:
                    cursor.execute('''
                    SELECT dr.id, dr.item_id, dr.card_id, dr.delivery_count, dr.delivery_times,
                           dr.enabled, dr.description, dr.user_id, dr.created_at, dr.updated_at,
                           c.name as card_name, c.type as card_type, dr.is_multi_spec, dr.spec_name, dr.spec_value
                    FROM delivery_rules dr
                    LEFT JOIN cards c ON dr.card_id = c.id
                    ORDER BY dr.created_at DESC
                    ''')
                
                rules = []
                for row in cursor.fetchall():
                    rules.append({
                        'id': row[0],
                        'item_id': row[1],
                        'card_id': row[2],
                        'delivery_count': row[3],
                        'delivery_times': row[4],
                        'enabled': bool(row[5]),
                        'description': row[6],
                        'user_id': row[7],
                        'created_at': row[8],
                        'updated_at': row[9],
                        'card_name': row[10],
                        'card_type': row[11],
                        'is_multi_spec': bool(row[12]) if row[12] is not None else False,
                        'spec_name': row[13],
                        'spec_value': row[14]
                    })
                
                return rules
            except Exception as e:
                logger.error(f"获取发货规则失败: {e}")
                return []

    def get_delivery_rule_by_id(self, rule_id: int, user_id: int = None) -> Optional[Dict[str, Any]]:
        """根据ID获取发货规则"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                if user_id is not None:
                    cursor.execute('''
                    SELECT dr.id, dr.item_id, dr.card_id, dr.delivery_count, dr.delivery_times,
                           dr.enabled, dr.description, dr.user_id, dr.created_at, dr.updated_at,
                           c.name as card_name, c.type as card_type, dr.is_multi_spec, dr.spec_name, dr.spec_value
                    FROM delivery_rules dr
                    LEFT JOIN cards c ON dr.card_id = c.id
                    WHERE dr.id = %s AND dr.user_id = %s
                    ''', (rule_id, user_id))
                else:
                    cursor.execute('''
                    SELECT dr.id, dr.item_id, dr.card_id, dr.delivery_count, dr.delivery_times,
                           dr.enabled, dr.description, dr.user_id, dr.created_at, dr.updated_at,
                           c.name as card_name, c.type as card_type, dr.is_multi_spec, dr.spec_name, dr.spec_value
                    FROM delivery_rules dr
                    LEFT JOIN cards c ON dr.card_id = c.id
                    WHERE dr.id = %s
                    ''', (rule_id,))
                
                row = cursor.fetchone()
                if row:
                    return {
                        'id': row[0],
                        'item_id': row[1],
                        'card_id': row[2],
                        'delivery_count': row[3],
                        'delivery_times': row[4],
                        'enabled': bool(row[5]),
                        'description': row[6],
                        'user_id': row[7],
                        'created_at': row[8],
                        'updated_at': row[9],
                        'card_name': row[10],
                        'card_type': row[11],
                        'is_multi_spec': bool(row[12]) if row[12] is not None else False,
                        'spec_name': row[13],
                        'spec_value': row[14]
                    }
                return None
            except Exception as e:
                logger.error(f"根据ID获取发货规则失败: {e}")
                return None

    def get_delivery_rules_by_item(self, item_id: str, user_id: int = None) -> List[Dict[str, Any]]:
        """获取指定商品的发货规则"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                if user_id is not None:
                    cursor.execute('''
                    SELECT dr.id, dr.item_id, dr.card_id, dr.delivery_count, dr.delivery_times,
                           dr.enabled, dr.description, dr.user_id, dr.created_at, dr.updated_at,
                           c.name as card_name, c.type as card_type, c.api_config,
                           c.text_content, c.data_content, c.image_url, c.enabled as card_enabled,
                           c.description as card_description, c.delay_seconds as card_delay_seconds,
                           c.is_multi_spec, c.spec_name, c.spec_value,
                           dr.is_multi_spec as rule_is_multi_spec, dr.spec_name as rule_spec_name, dr.spec_value as rule_spec_value
                    FROM delivery_rules dr
                    LEFT JOIN cards c ON dr.card_id = c.id
                    WHERE dr.item_id = %s AND dr.user_id = %s AND dr.enabled = 1 
                    AND (c.enabled = 1 OR c.enabled IS NULL)
                    ORDER BY dr.created_at DESC
                    ''', (item_id, user_id))
                else:
                    cursor.execute('''
                    SELECT dr.id, dr.item_id, dr.card_id, dr.delivery_count, dr.delivery_times,
                           dr.enabled, dr.description, dr.user_id, dr.created_at, dr.updated_at,
                           c.name as card_name, c.type as card_type, c.api_config,
                           c.text_content, c.data_content, c.image_url, c.enabled as card_enabled,
                           c.description as card_description, c.delay_seconds as card_delay_seconds,
                           c.is_multi_spec, c.spec_name, c.spec_value,
                           dr.is_multi_spec as rule_is_multi_spec, dr.spec_name as rule_spec_name, dr.spec_value as rule_spec_value
                    FROM delivery_rules dr
                    LEFT JOIN cards c ON dr.card_id = c.id
                    WHERE dr.item_id = %s AND dr.enabled = 1 
                    AND (c.enabled = 1 OR c.enabled IS NULL)
                    ORDER BY dr.created_at DESC
                    ''', (item_id,))
                
                rules = []
                for row in cursor.fetchall():
                    # 解析api_config JSON字符串
                    api_config = row[12]
                    if api_config:
                        try:
                            import json
                            api_config = json.loads(api_config)
                        except (json.JSONDecodeError, TypeError):
                            # 如果解析失败，保持原始字符串
                            pass
                    
                    rules.append({
                        'id': row[0],
                        'item_id': row[1],
                        'card_id': row[2],
                        'delivery_count': row[3],
                        'delivery_times': row[4],
                        'enabled': bool(row[5]),
                        'description': row[6],
                        'user_id': row[7],
                        'created_at': row[8],
                        'updated_at': row[9],
                        'card_name': row[10],
                        'card_type': row[11],
                        'api_config': api_config,
                        'text_content': row[13],
                        'data_content': row[14],
                        'image_url': row[15],
                        'card_enabled': bool(row[16]) if row[16] is not None else None,
                        'card_description': row[17],
                        'card_delay_seconds': row[18] or 0,
                        'is_multi_spec': bool(row[19]) if row[19] is not None else False,
                        'spec_name': row[20],
                        'spec_value': row[21],
                        'rule_is_multi_spec': bool(row[22]) if row[22] is not None else False,
                        'rule_spec_name': row[23],
                        'rule_spec_value': row[24]
                    })
                
                return rules
            except Exception as e:
                logger.error(f"获取商品发货规则失败: {e}")
                return []

    def get_delivery_rules_by_item_id(self, item_id: str, user_id: int = None) -> List[Dict[str, Any]]:
        """获取指定商品ID的发货规则（别名方法）"""
        return self.get_delivery_rules_by_item(item_id, user_id)

    def get_all_delivery_rules(self, user_id: int = None) -> List[Dict[str, Any]]:
        """获取所有发货规则（别名方法）"""
        return self.get_delivery_rules(user_id)

    def update_delivery_rule(self, rule_id: int, item_id: str = None, card_id: int = None,
                           delivery_count: int = None, description: str = None,
                           enabled: bool = None, generate_image: bool = None,
                           is_multi_spec: bool = None, spec_name: str = None, spec_value: str = None) -> bool:
        """更新发货规则"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 构建动态更新SQL
                update_fields = []
                params = []
                
                if item_id is not None:
                    update_fields.append("item_id = %s")
                    params.append(item_id)
                if card_id is not None:
                    update_fields.append("card_id = %s")
                    params.append(card_id)
                if delivery_count is not None:
                    update_fields.append("delivery_count = %s")
                    params.append(delivery_count)
                if description is not None:
                    update_fields.append("description = %s")
                    params.append(description)
                if enabled is not None:
                    update_fields.append("enabled = %s")
                    params.append(enabled)
                if generate_image is not None:
                    update_fields.append("generate_image = %s")
                    params.append(generate_image)
                if is_multi_spec is not None:
                    update_fields.append("is_multi_spec = %s")
                    params.append(is_multi_spec)
                if spec_name is not None:
                    update_fields.append("spec_name = %s")
                    params.append(spec_name)
                if spec_value is not None:
                    update_fields.append("spec_value = %s")
                    params.append(spec_value)
                
                if update_fields:
                    update_fields.append("updated_at = CURRENT_TIMESTAMP")
                    params.append(rule_id)
                    
                    sql = f"UPDATE delivery_rules SET {', '.join(update_fields)} WHERE id = %s"
                    cursor.execute(sql, params)
                    
                    if cursor.rowcount > 0:
                        self.conn.commit()
                        logger.info(f"发货规则更新成功: {rule_id}")
                        return True
                    else:
                        logger.warning(f"发货规则不存在: {rule_id}")
                        return False
                else:
                    logger.warning("没有提供要更新的字段")
                    return False
                    
            except Exception as e:
                logger.error(f"更新发货规则失败: {e}")
                self.conn.rollback()
                return False

    def delete_delivery_rule(self, rule_id: int, user_id: int = None) -> bool:
        """删除发货规则"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                if user_id is not None:
                    cursor.execute('DELETE FROM delivery_rules WHERE id = %s AND user_id = %s', (rule_id, user_id))
                else:
                    cursor.execute('DELETE FROM delivery_rules WHERE id = %s', (rule_id,))
                
                if cursor.rowcount > 0:
                    self.conn.commit()
                    logger.info(f"发货规则删除成功: {rule_id}")
                    return True
                else:
                    logger.warning(f"发货规则不存在或无权删除: {rule_id}")
                    return False
            except Exception as e:
                logger.error(f"删除发货规则失败: {e}")
                self.conn.rollback()
                return False

    def increment_delivery_times(self, rule_id: int) -> bool:
        """增加发货次数"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                UPDATE delivery_rules 
                SET delivery_times = delivery_times + 1, updated_at = CURRENT_TIMESTAMP
                WHERE id = %s
                ''', (rule_id,))
                
                if cursor.rowcount > 0:
                    self.conn.commit()
                    logger.debug(f"发货次数增加成功: 规则{rule_id}")
                    return True
                else:
                    logger.warning(f"发货规则不存在: {rule_id}")
                    return False
            except Exception as e:
                logger.error(f"增加发货次数失败: {e}")
                self.conn.rollback()
                return False

    def get_delivery_rules_stats(self, user_id: int = None) -> Dict[str, int]:
        """获取发货规则统计"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                base_where = "WHERE user_id = %s" if user_id else ""
                params = [user_id] if user_id else []
                
                # 总规则数
                cursor.execute(f'SELECT COUNT(*) FROM delivery_rules {base_where}', params)
                total_rules = cursor.fetchone()[0]
                
                # 启用规则数
                enabled_where = f"{base_where} {'AND' if base_where else 'WHERE'} enabled = TRUE"
                cursor.execute(f'SELECT COUNT(*) FROM delivery_rules {enabled_where}', params)
                enabled_rules = cursor.fetchone()[0]
                
                # 今日发货次数
                today_where = f"{base_where} {'AND' if base_where else 'WHERE'} DATE(updated_at) = CURDATE()"
                cursor.execute(f'SELECT SUM(delivery_times) FROM delivery_rules {today_where}', params)
                today_deliveries = cursor.fetchone()[0] or 0
                
                return {
                    'total_rules': total_rules,
                    'enabled_rules': enabled_rules,
                    'today_deliveries': today_deliveries
                }
                
            except Exception as e:
                logger.error(f"获取发货规则统计失败: {e}")
                return {
                    'total_rules': 0,
                    'enabled_rules': 0,
                    'today_deliveries': 0
                }

    # -------------------- 用户设置管理 --------------------
    def get_user_settings(self, user_id: int) -> Dict[str, Any]:
        """获取用户的所有设置"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT `key`, value, description, created_at, updated_at
                FROM user_settings 
                WHERE user_id = %s
                ORDER BY `key`
                ''', (user_id,))
                
                settings = {}
                for row in cursor.fetchall():
                    settings[row[0]] = {
                        'value': row[1],
                        'description': row[2],
                        'created_at': row[3],
                        'updated_at': row[4]
                    }
                
                return settings
            except Exception as e:
                logger.error(f"获取用户设置失败: {e}")
                return {}

    def get_user_setting(self, user_id: int, key: str) -> Optional[Dict[str, Any]]:
        """获取用户的特定设置"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT value, description, created_at, updated_at
                FROM user_settings 
                WHERE user_id = %s AND `key` = %s
                ''', (user_id, key))
                
                result = cursor.fetchone()
                if result:
                    return {
                        'key': key,
                        'value': result[0],
                        'description': result[1],
                        'created_at': result[2],
                        'updated_at': result[3]
                    }
                return None
            except Exception as e:
                logger.error(f"获取用户设置失败: {e}")
                return None

    def set_user_setting(self, user_id: int, key: str, value: str, description: str = None) -> bool:
        """设置用户设置"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                INSERT INTO user_settings (user_id, `key`, value, description)
                VALUES (%s, %s, %s, %s)
                ON DUPLICATE KEY UPDATE
                value = VALUES(value),
                description = VALUES(description),
                updated_at = CURRENT_TIMESTAMP
                ''', (user_id, key, value, description))
                
                self.conn.commit()
                logger.info(f"用户设置保存成功: 用户{user_id} - {key} = {value}")
                return True
            except Exception as e:
                logger.error(f"设置用户设置失败: {e}")
                self.conn.rollback()
                return False

    def delete_user_setting(self, user_id: int, key: str) -> bool:
        """删除用户设置"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                DELETE FROM user_settings 
                WHERE user_id = %s AND `key` = %s
                ''', (user_id, key))
                
                if cursor.rowcount > 0:
                    self.conn.commit()
                    logger.info(f"用户设置删除成功: 用户{user_id} - {key}")
                    return True
                else:
                    logger.warning(f"用户设置不存在: 用户{user_id} - {key}")
                    return False
            except Exception as e:
                logger.error(f"删除用户设置失败: {e}")
                self.conn.rollback()
                return False

    def clear_user_settings(self, user_id: int) -> bool:
        """清空用户的所有设置"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                DELETE FROM user_settings WHERE user_id = %s
                ''', (user_id,))
                
                deleted_count = cursor.rowcount
                self.conn.commit()
                logger.info(f"清空用户设置: 用户{user_id}, 删除 {deleted_count} 条")
                return True
            except Exception as e:
                logger.error(f"清空用户设置失败: {e}")
                self.conn.rollback()
                return False

    # -------------------- 门店词库管理 --------------------
    def create_keyword_library(self, user_id: int, name: str, description: str = None) -> int:
        """创建门店词库"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                INSERT INTO keyword_libraries (user_id, name, description)
                VALUES (%s, %s, %s)
                ''', (user_id, name, description))
                
                library_id = cursor.lastrowid
                self.conn.commit()
                logger.info(f"门店词库创建成功: {name} (ID: {library_id})")
                return library_id
            except Exception as e:
                logger.error(f"创建门店词库失败: {e}")
                self.conn.rollback()
                raise

    def get_keyword_libraries(self, user_id: int = None) -> List[Dict[str, Any]]:
        """获取门店词库列表（全系统共享）"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 门店词库改为全系统共享，所有用户都能看到所有词库
                cursor.execute('''
                SELECT 
                    kl.id, 
                    kl.user_id, 
                    kl.name, 
                    kl.description,
                    (SELECT COUNT(*) FROM keyword_library_items WHERE library_id = kl.id) as keywords_count,
                    COALESCE(kl.share_count, 0) as share_count,
                    kl.created_at, 
                    kl.updated_at,
                    u.username as creator_name
                FROM keyword_libraries kl
                LEFT JOIN users u ON kl.user_id = u.id
                ORDER BY kl.created_at DESC
                ''')

                libraries = []
                for row in cursor.fetchall():
                    libraries.append({
                        'id': row[0],
                        'user_id': row[1],
                        'name': row[2],
                        'description': row[3],
                        'keywords_count': row[4],
                        'share_count': row[5],
                        'created_at': row[6],
                        'updated_at': row[7]
                    })

                return libraries
            except Exception as e:
                logger.error(f"获取门店词库列表失败: {e}")
                return []

    def get_keyword_library_by_id(self, library_id: int, user_id: int = None) -> Optional[Dict[str, Any]]:
        """根据ID获取门店词库"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                if user_id is not None:
                    cursor.execute('''
                    SELECT id, user_id, name, description, share_count, created_at, updated_at
                    FROM keyword_libraries
                    WHERE id = %s AND user_id = %s
                    ''', (library_id, user_id))
                else:
                    cursor.execute('''
                    SELECT id, user_id, name, description, share_count, created_at, updated_at
                    FROM keyword_libraries
                    WHERE id = %s
                    ''', (library_id,))

                row = cursor.fetchone()
                if row:
                    return {
                        'id': row[0],
                        'user_id': row[1],
                        'name': row[2],
                        'description': row[3],
                        'share_count': row[4],
                        'created_at': row[5],
                        'updated_at': row[6]
                    }
                return None
            except Exception as e:
                logger.error(f"根据ID获取门店词库失败: {e}")
                return None

    def update_keyword_library(self, library_id: int, name: str = None, description: str = None) -> bool:
        """更新门店词库"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 构建动态更新SQL
                update_fields = []
                params = []
                
                if name is not None:
                    update_fields.append("name = %s")
                    params.append(name)
                if description is not None:
                    update_fields.append("description = %s")
                    params.append(description)
                
                if update_fields:
                    update_fields.append("updated_at = CURRENT_TIMESTAMP")
                    params.append(library_id)
                    
                    sql = f"UPDATE keyword_libraries SET {', '.join(update_fields)} WHERE id = %s"
                    cursor.execute(sql, params)
                    
                    if cursor.rowcount > 0:
                        self.conn.commit()
                        logger.info(f"门店词库更新成功: {library_id}")
                        return True
                    else:
                        logger.warning(f"关键词库不存在: {library_id}")
                        return False
                else:
                    logger.warning("没有提供要更新的字段")
                    return False
                    
            except Exception as e:
                logger.error(f"更新门店词库失败: {e}")
                self.conn.rollback()
                return False

    def delete_keyword_library(self, library_id: int, user_id: int = None) -> bool:
        """删除门店词库"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                if user_id is not None:
                    cursor.execute('DELETE FROM keyword_libraries WHERE id = %s AND user_id = %s', (library_id, user_id))
                else:
                    cursor.execute('DELETE FROM keyword_libraries WHERE id = %s', (library_id,))
                
                if cursor.rowcount > 0:
                    self.conn.commit()
                    logger.info(f"门店词库删除成功: {library_id}")
                    return True
                else:
                    logger.warning(f"门店词库不存在或无权删除: {library_id}")
                    return False
            except Exception as e:
                logger.error(f"删除门店词库失败: {e}")
                self.conn.rollback()
                return False

    def add_keyword_to_library(self, library_id: int, keyword: str, reply: str, 
                             item_id: str = None, keyword_type: str = 'text', image_url: str = None) -> bool:
        """向门店词库添加关键词"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                INSERT INTO keyword_library_items (library_id, keyword, reply, item_id, type, image_url)
                VALUES (%s, %s, %s, %s, %s, %s)
                ''', (library_id, keyword, reply, item_id, keyword_type, image_url))
                
                self.conn.commit()
                logger.info(f"关键词添加到库成功: 库{library_id} - {keyword}")
                return True
            except Exception as e:
                logger.error(f"向门店词库添加关键词失败: {e}")
                self.conn.rollback()
                return False

    def get_keyword_library_items(self, library_id: int) -> List[Dict[str, Any]]:
        """获取门店词库中的关键词"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT id, library_id, keyword, reply, item_id, type, image_url, created_at
                FROM keyword_library_items
                WHERE library_id = %s
                ORDER BY created_at DESC
                ''', (library_id,))
                
                items = []
                for row in cursor.fetchall():
                    items.append({
                        'id': row[0],
                        'library_id': row[1],
                        'keyword': row[2],
                        'reply': row[3],
                        'item_id': row[4],
                        'type': row[5] or 'text',
                        'image_url': row[6],
                        'created_at': row[7]
                    })
                
                return items
            except Exception as e:
                logger.error(f"获取门店词库项目失败: {e}")
                return []

    def delete_keyword_library_item(self, item_id: int) -> bool:
        """删除门店词库中的关键词"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('DELETE FROM keyword_library_items WHERE id = %s', (item_id,))
                
                if cursor.rowcount > 0:
                    self.conn.commit()
                    logger.info(f"门店词库项目删除成功: {item_id}")
                    return True
                else:
                    logger.warning(f"门店词库项目不存在: {item_id}")
                    return False
            except Exception as e:
                logger.error(f"删除门店词库项目失败: {e}")
                self.conn.rollback()
                return False

    def increment_library_share_count(self, library_id: int) -> bool:
        """增加门店词库分享次数"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                UPDATE keyword_libraries 
                SET share_count = share_count + 1, updated_at = CURRENT_TIMESTAMP
                WHERE id = %s
                ''', (library_id,))
                
                if cursor.rowcount > 0:
                    self.conn.commit()
                    logger.debug(f"关键词库分享次数增加: {library_id}")
                    return True
                else:
                    logger.warning(f"关键词库不存在: {library_id}")
                    return False
            except Exception as e:
                logger.error(f"增加关键词库分享次数失败: {e}")
                self.conn.rollback()
                return False

    # -------------------- 通用关键词管理 --------------------
    def get_global_keywords(self, user_id: int = None) -> List[Dict[str, Any]]:
        """获取通用关键词"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                if user_id is not None:
                    cursor.execute('''
                    SELECT id, keyword, reply, type, image_url, enabled, is_fuzzy, created_at, updated_at
                    FROM global_keywords
                    WHERE user_id = %s
                    ORDER BY created_at DESC
                    ''', (user_id,))
                else:
                    cursor.execute('''
                    SELECT id, keyword, reply, type, image_url, enabled, is_fuzzy, created_at, updated_at
                    FROM global_keywords
                    ORDER BY created_at DESC
                    ''')
                
                keywords = []
                for row in cursor.fetchall():
                    keywords.append({
                        'id': row[0],
                        'keyword': row[1],
                        'reply': row[2],
                        'type': row[3] or 'text',
                        'image_url': row[4],
                        'enabled': bool(row[5]),
                        'is_fuzzy': bool(row[6]),
                        'created_at': row[7],
                        'updated_at': row[8]
                    })
                
                return keywords
            except Exception as e:
                logger.error(f"获取通用关键词失败: {e}")
                return []

    def get_enabled_global_keywords(self, user_id: int = None) -> List[Dict[str, Any]]:
        """获取启用的通用关键词"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                if user_id is not None:
                    cursor.execute('''
                    SELECT id, keyword, reply, type, image_url, is_fuzzy, created_at, updated_at
                    FROM global_keywords
                    WHERE enabled = TRUE AND user_id = %s
                    ORDER BY created_at DESC
                    ''', (user_id,))
                else:
                    cursor.execute('''
                    SELECT id, keyword, reply, type, image_url, is_fuzzy, created_at, updated_at
                    FROM global_keywords
                    WHERE enabled = TRUE
                    ORDER BY created_at DESC
                    ''')
                
                keywords = []
                for row in cursor.fetchall():
                    keywords.append({
                        'id': row[0],
                        'keyword': row[1],
                        'reply': row[2],
                        'type': row[3] or 'text',
                        'image_url': row[4],
                        'is_fuzzy': bool(row[5]),
                        'created_at': row[6],
                        'updated_at': row[7]
                    })
                
                return keywords
            except Exception as e:
                logger.error(f"获取启用的通用关键词失败: {e}")
                return []

    def add_global_keyword(self, keyword: str, reply: str, keyword_type: str = 'text', 
                         image_url: str = None, enabled: bool = True, is_fuzzy: bool = False, user_id: int = None) -> bool:
        """添加通用关键词"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                INSERT INTO global_keywords (user_id, keyword, reply, type, image_url, enabled, is_fuzzy)
                VALUES (%s, %s, %s, %s, %s, %s, %s)
                ''', (user_id, keyword, reply, keyword_type, image_url, enabled, is_fuzzy))
                
                self.conn.commit()
                logger.info(f"通用关键词添加成功: {keyword} (用户ID: {user_id})")
                return True
            except pymysql.IntegrityError as ie:
                if "Duplicate entry" in str(ie):
                    logger.error(f"通用关键词已存在: {keyword} (用户ID: {user_id})")
                else:
                    logger.error(f"通用关键词添加失败: {ie}")
                self.conn.rollback()
                return False
            except Exception as e:
                logger.error(f"添加通用关键词失败: {e}")
                self.conn.rollback()
                return False

    def update_global_keyword(self, keyword_id: int, keyword: str = None, reply: str = None,
                            keyword_type: str = None, image_url: str = None, enabled: bool = None, is_fuzzy: bool = None) -> bool:
        """更新通用关键词"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 构建动态更新SQL
                update_fields = []
                params = []
                
                if keyword is not None:
                    update_fields.append("keyword = %s")
                    params.append(keyword)
                if reply is not None:
                    update_fields.append("reply = %s")
                    params.append(reply)
                if keyword_type is not None:
                    update_fields.append("type = %s")
                    params.append(keyword_type)
                if image_url is not None:
                    update_fields.append("image_url = %s")
                    params.append(image_url)
                if enabled is not None:
                    update_fields.append("enabled = %s")
                    params.append(enabled)
                if is_fuzzy is not None:
                    update_fields.append("is_fuzzy = %s")
                    params.append(is_fuzzy)
                
                if update_fields:
                    update_fields.append("updated_at = CURRENT_TIMESTAMP")
                    params.append(keyword_id)
                    
                    sql = f"UPDATE global_keywords SET {', '.join(update_fields)} WHERE id = %s"
                    cursor.execute(sql, params)
                    
                    if cursor.rowcount > 0:
                        self.conn.commit()
                        logger.info(f"通用关键词更新成功: {keyword_id}")
                        return True
                    else:
                        logger.warning(f"通用关键词不存在: {keyword_id}")
                        return False
                else:
                    logger.warning("没有提供要更新的字段")
                    return False
                    
            except Exception as e:
                logger.error(f"更新通用关键词失败: {e}")
                self.conn.rollback()
                return False

    def delete_global_keyword(self, keyword_id: int) -> bool:
        """删除通用关键词"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('DELETE FROM global_keywords WHERE id = %s', (keyword_id,))
                
                if cursor.rowcount > 0:
                    self.conn.commit()
                    logger.info(f"通用关键词删除成功: {keyword_id}")
                    return True
                else:
                    logger.warning(f"通用关键词不存在: {keyword_id}")
                    return False
            except Exception as e:
                logger.error(f"删除通用关键词失败: {e}")
                self.conn.rollback()
                return False

    def get_global_keyword_by_id(self, keyword_id: int) -> Optional[Dict[str, Any]]:
        """根据ID获取通用关键词"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT id, keyword, reply, type, image_url, enabled, is_fuzzy, created_at, updated_at
                FROM global_keywords
                WHERE id = %s
                ''', (keyword_id,))
                
                row = cursor.fetchone()
                if row:
                    return {
                        'id': row[0],
                        'keyword': row[1],
                        'reply': row[2],
                        'type': row[3] or 'text',
                        'image_url': row[4],
                        'enabled': bool(row[5]),
                        'is_fuzzy': bool(row[6]),
                        'created_at': row[7],
                        'updated_at': row[8]
                    }
                return None
            except Exception as e:
                logger.error(f"根据ID获取通用关键词失败: {e}")
                return None

    def search_global_keywords(self, query: str, user_id: int = None) -> List[Dict[str, Any]]:
        """搜索通用关键词"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                search_param = f"%{query}%"
                if user_id is not None:
                    cursor.execute('''
                    SELECT id, keyword, reply, type, image_url, enabled, is_fuzzy, created_at, updated_at
                    FROM global_keywords
                    WHERE (keyword LIKE %s OR reply LIKE %s) AND user_id = %s
                    ORDER BY created_at DESC
                    ''', (search_param, search_param, user_id))
                else:
                    cursor.execute('''
                    SELECT id, keyword, reply, type, image_url, enabled, is_fuzzy, created_at, updated_at
                    FROM global_keywords
                    WHERE keyword LIKE %s OR reply LIKE %s
                    ORDER BY created_at DESC
                    ''', (search_param, search_param))
                
                keywords = []
                for row in cursor.fetchall():
                    keywords.append({
                        'id': row[0],
                        'keyword': row[1],
                        'reply': row[2],
                        'type': row[3] or 'text',
                        'image_url': row[4],
                        'enabled': bool(row[5]),
                        'is_fuzzy': bool(row[6]),
                        'created_at': row[7],
                        'updated_at': row[8]
                    })
                
                return keywords
            except Exception as e:
                logger.error(f"搜索通用关键词失败: {e}")
                return []

    def batch_add_global_keywords(self, keywords: List[Tuple[str, str, str]]) -> int:
        """批量添加通用关键词"""
        if not keywords:
            return 0

        success_count = 0
        with self.lock:
            try:
                cursor = self.get_connection().cursor()

                for keyword, reply, keyword_type in keywords:
                    try:
                        cursor.execute('''
                        INSERT INTO global_keywords (keyword, reply, type, enabled)
                        VALUES (%s, %s, %s, TRUE)
                        ''', (keyword, reply, keyword_type or 'text'))
                        success_count += 1
                    except pymysql.IntegrityError:
                        # 如果遇到唯一约束冲突，跳过这个关键词
                        continue
                    except Exception as e:
                        logger.error(f"批量添加通用关键词项失败 {keyword}: {e}")
                        continue

                self.conn.commit()
                logger.info(f"批量添加通用关键词完成: {success_count}/{len(keywords)} 个")
                return success_count

            except Exception as e:
                logger.error(f"批量添加通用关键词失败: {e}")
                self.conn.rollback()
                return success_count

    def add_keywords_to_library(self, library_id: int, keywords: List[Dict[str, Any]]) -> bool:
        """向关键词库添加关键词"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 批量插入关键词（性能优化）
                if keywords:
                    # 准备批量插入数据
                    batch_data = []
                    for keyword_data in keywords:
                        batch_data.append((
                            library_id,
                            keyword_data['keyword'],
                            keyword_data['reply'],
                            keyword_data.get('type', 'text'),
                            keyword_data.get('image_url')
                        ))
                    
                    # 使用executemany进行批量插入
                    start_time = time.time()
                    cursor.executemany(
                        "INSERT INTO keyword_library_items (library_id, keyword, reply, type, image_url) VALUES (%s, %s, %s, %s, %s)",
                        batch_data
                    )
                    insert_time = time.time() - start_time
                    logger.debug(f"批量插入 {len(batch_data)} 个关键词耗时: {insert_time:.2f}s")
                
                # 更新关键词数量
                self._execute_sql(cursor,
                    "UPDATE keyword_libraries SET updated_at = NOW() WHERE id = %s",
                    (library_id,))
                
                self.conn.commit()
                logger.info(f"关键词库批量添加关键词成功: library_id={library_id}, count={len(keywords)}")
                return True
            except Exception as e:
                logger.error(f"向关键词库添加关键词失败: {e}")
                self.conn.rollback()
                return False

    def parse_order_message(self, message_data: dict) -> Optional[dict]:
        """解析订单消息，提取订单信息"""
        try:
            # 从消息中提取订单ID
            order_id = None
            order_status = "未知"
            buyer_message = ""
            
            # 添加调试日志
            logger.debug(f"开始解析订单消息")
            
            # 解析消息内容 - 支持多种消息格式
            # 格式1: 标准聊天消息格式
            if isinstance(message_data, dict) and '6' in message_data and isinstance(message_data['6'], dict) and '3' in message_data['6']:
                logger.debug("检测到格式1: 标准聊天消息格式")
                content = message_data['6']['3']
                if isinstance(content, dict):
                    if '2' in content and isinstance(content['2'], str):
                        buyer_message = content['2']
                        logger.debug(f"提取到买家消息: {buyer_message}")
                    
                    if '5' in content and isinstance(content['5'], str):
                        # 解析JSON内容
                        try:
                            import json
                            json_content = json.loads(content['5'])
                            logger.debug(f"解析JSON内容成功")
                            if 'dxCard' in json_content and 'item' in json_content['dxCard']:
                                card_item = json_content['dxCard']['item']
                                if 'targetUrl' in card_item.get('main', {}):
                                    target_url = card_item['main']['targetUrl']
                                    logger.debug(f"找到targetUrl: {target_url}")
                                    # 从URL中提取订单ID
                                    import re
                                    order_match = re.search(r'orderId=(\d+)', target_url)
                                    if order_match:
                                        order_id = order_match.group(1)
                                        logger.debug(f"从targetUrl提取到订单ID: {order_id}")
                        except Exception as e:
                            logger.debug(f"解析JSON内容失败: {e}")
            
            # 如果找到了订单ID，返回解析结果
            if order_id:
                return {
                    'order_id': order_id,
                    'status': order_status,
                    'buyer_message': buyer_message
                }
            else:
                logger.debug("未找到订单ID")
                return None
                
        except Exception as e:
            logger.error(f"解析订单消息失败: {e}")
            return None

    def cancel_delayed_guide_tasks(self, cookie_id: str, user_id: str, chat_id: str) -> bool:
        """取消指定对话的所有待发送延迟引导任务"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                import datetime
                current_time = datetime.datetime.now()
                
                self._execute_sql(cursor, '''
                UPDATE delayed_guide_tasks 
                SET status = 'cancelled', cancelled_at = %s
                WHERE cookie_id = %s AND user_id = %s AND chat_id = %s AND status = 'pending'
                ''', (current_time, cookie_id, user_id, chat_id))
                
                affected_rows = cursor.rowcount
                self.conn.commit()
                if affected_rows > 0:
                    logger.info(f"取消 {affected_rows} 个延迟引导任务: 账号:{cookie_id}, 用户:{user_id}, 对话:{chat_id}")
                
                return affected_rows > 0
                
            except Exception as e:
                logger.error(f"取消延迟引导任务失败: {e}")
                self.conn.rollback()
                return False

    def add_delayed_guide_task(self, cookie_id: str, user_id: str, chat_id: str, 
                              item_id: str = None, delay_seconds: int = 60) -> int:
        """添加延迟引导话术任务"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                # 计算延迟发送时间
                import datetime
                scheduled_time = datetime.datetime.now() + datetime.timedelta(seconds=delay_seconds)
                trigger_time = datetime.datetime.now()
                
                self._execute_sql(cursor, '''
                INSERT INTO delayed_guide_tasks 
                (cookie_id, user_id, chat_id, item_id, trigger_time, scheduled_time, status)
                VALUES (%s, %s, %s, %s, %s, %s, 'pending')
                ''', (cookie_id, user_id, chat_id, item_id, trigger_time, scheduled_time))
                
                task_id = cursor.lastrowid
                self.conn.commit()
                logger.info(f"添加延迟引导任务: {task_id}, 账号: {cookie_id}, 用户: {user_id}")
                return task_id
                
            except Exception as e:
                logger.error(f"添加延迟引导任务失败: {e}")
                self.conn.rollback()
                return 0

    def mark_delayed_guide_task_sent(self, task_id: int) -> bool:
        """标记延迟引导任务为已发送"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                import datetime
                current_time = datetime.datetime.now()
                
                self._execute_sql(cursor, '''
                UPDATE delayed_guide_tasks 
                SET status = 'sent', sent_at = %s
                WHERE id = %s
                ''', (current_time, task_id))
                
                self.conn.commit()
                return cursor.rowcount > 0
                
            except Exception as e:
                logger.error(f"标记延迟引导任务为已发送失败: {e}")
                self.conn.rollback()
                return False

    def is_chat_first_reply(self, cookie_id: str, chat_id: str) -> bool:
        """检查是否为对话的首次回复（别名方法，调用 has_chat_first_replied）"""
        return not self.has_chat_first_replied(cookie_id, chat_id)

    def get_delivery_rules_by_item_and_spec(self, item_id: str, spec_name: str = None, spec_value: str = None, user_id: int = None) -> List[Dict[str, Any]]:
        """根据商品ID和规格信息获取匹配的发货规则（支持多规格）"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()

                # 优先匹配：商品ID+规格名称+规格值
                if spec_name and spec_value:
                    if user_id is not None:
                        cursor.execute('''
                        SELECT dr.id, dr.item_id, dr.card_id, dr.delivery_count, dr.delivery_times,
                               dr.enabled, dr.description, dr.user_id, dr.created_at, dr.updated_at,
                               c.name as card_name, c.type as card_type, c.api_config,
                               c.text_content, c.data_content, c.image_url, c.enabled as card_enabled,
                               c.description as card_description, c.delay_seconds as card_delay_seconds,
                               c.is_multi_spec, c.spec_name, c.spec_value,
                               dr.is_multi_spec as rule_is_multi_spec, dr.spec_name as rule_spec_name, dr.spec_value as rule_spec_value
                        FROM delivery_rules dr
                        LEFT JOIN cards c ON dr.card_id = c.id
                        WHERE dr.item_id = %s AND dr.user_id = %s AND dr.enabled = 1 
                        AND (c.enabled = 1 OR c.enabled IS NULL)
                        AND dr.is_multi_spec = 1 AND dr.spec_name = %s AND dr.spec_value = %s
                        ORDER BY dr.created_at DESC
                        ''', (item_id, user_id, spec_name, spec_value))
                    else:
                        cursor.execute('''
                        SELECT dr.id, dr.item_id, dr.card_id, dr.delivery_count, dr.delivery_times,
                               dr.enabled, dr.description, dr.user_id, dr.created_at, dr.updated_at,
                               c.name as card_name, c.type as card_type, c.api_config,
                               c.text_content, c.data_content, c.image_url, c.enabled as card_enabled,
                               c.description as card_description, c.delay_seconds as card_delay_seconds,
                               c.is_multi_spec, c.spec_name, c.spec_value,
                               dr.is_multi_spec as rule_is_multi_spec, dr.spec_name as rule_spec_name, dr.spec_value as rule_spec_value
                        FROM delivery_rules dr
                        LEFT JOIN cards c ON dr.card_id = c.id
                        WHERE dr.item_id = %s AND dr.enabled = 1 
                        AND (c.enabled = 1 OR c.enabled IS NULL)
                        AND dr.is_multi_spec = 1 AND dr.spec_name = %s AND dr.spec_value = %s
                        ORDER BY dr.created_at DESC
                        ''', (item_id, spec_name, spec_value))

                    rules = []
                    for row in cursor.fetchall():
                        # 解析api_config JSON字符串
                        api_config = row[12]
                        if api_config:
                            try:
                                import json
                                api_config = json.loads(api_config)
                            except (json.JSONDecodeError, TypeError):
                                # 如果解析失败，保持原始字符串
                                pass

                        rules.append({
                            'id': row[0],
                            'item_id': row[1],
                            'card_id': row[2],
                            'delivery_count': row[3],
                            'delivery_times': row[4],
                            'enabled': bool(row[5]),
                            'description': row[6],
                            'user_id': row[7],
                            'created_at': row[8],
                            'updated_at': row[9],
                            'card_name': row[10],
                            'card_type': row[11],
                            'api_config': api_config,
                            'text_content': row[13],
                            'data_content': row[14],
                            'image_url': row[15],
                            'card_enabled': bool(row[16]) if row[16] is not None else None,
                            'card_description': row[17],
                            'card_delay_seconds': row[18] or 0,
                            'is_multi_spec': bool(row[19]),
                            'spec_name': row[20],
                            'spec_value': row[21],
                            'rule_is_multi_spec': bool(row[22]) if row[22] is not None else False,
                            'rule_spec_name': row[23],
                            'rule_spec_value': row[24]
                        })

                    if rules:
                        logger.info(f"找到多规格匹配规则: {item_id} - {spec_name}:{spec_value}")
                        return rules

                return []

            except Exception as e:
                logger.error(f"获取商品多规格发货规则失败: {e}")
                return []

    def get_delivery_rules_by_keyword_and_spec(self, keyword: str, spec_name: str = None, spec_value: str = None) -> List[Dict[str, Any]]:
        """根据关键字和规格信息获取匹配的发货规则（支持多规格）"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()

                # 优先匹配：卡券名称+规格名称+规格值
                if spec_name and spec_value:
                    cursor.execute('''
                    SELECT dr.id, dr.keyword, dr.card_id, dr.delivery_count, dr.enabled,
                           dr.description, dr.delivery_times,
                           c.name as card_name, c.type as card_type, c.api_config,
                           c.text_content, c.data_content, c.enabled as card_enabled,
                           c.description as card_description, c.delay_seconds as card_delay_seconds,
                           c.is_multi_spec, c.spec_name, c.spec_value
                    FROM delivery_rules dr
                    LEFT JOIN cards c ON dr.card_id = c.id
                    WHERE dr.enabled = 1 AND c.enabled = 1
                    AND (%s LIKE CONCAT('%%', dr.keyword, '%%') OR dr.keyword LIKE CONCAT('%%', %s, '%%'))
                    AND c.is_multi_spec = 1 AND c.spec_name = %s AND c.spec_value = %s
                    ORDER BY
                        CASE
                            WHEN %s LIKE CONCAT('%%', dr.keyword, '%%') THEN LENGTH(dr.keyword)
                            ELSE LENGTH(dr.keyword) / 2
                        END DESC,
                        dr.delivery_times ASC
                    ''', (keyword, keyword, spec_name, spec_value, keyword))

                    rules = []
                    for row in cursor.fetchall():
                        # 解析api_config JSON字符串
                        api_config = row[9]
                        if api_config:
                            try:
                                import json
                                api_config = json.loads(api_config)
                            except (json.JSONDecodeError, TypeError):
                                # 如果解析失败，保持原始字符串
                                pass

                        rules.append({
                            'id': row[0],
                            'keyword': row[1],
                            'card_id': row[2],
                            'delivery_count': row[3],
                            'enabled': bool(row[4]),
                            'description': row[5],
                            'delivery_times': row[6] or 0,
                            'card_name': row[7],
                            'card_type': row[8],
                            'api_config': api_config,
                            'text_content': row[10],
                            'data_content': row[11],
                            'card_enabled': bool(row[12]),
                            'card_description': row[13],
                            'card_delay_seconds': row[14] or 0,
                            'is_multi_spec': bool(row[15]),
                            'spec_name': row[16],
                            'spec_value': row[17]
                        })

                    if rules:
                        logger.info(f"找到多规格匹配规则: {keyword} - {spec_name}:{spec_value}")
                        return rules

                # 兜底匹配：仅卡券名称
                cursor.execute('''
                SELECT dr.id, dr.keyword, dr.card_id, dr.delivery_count, dr.enabled,
                       dr.description, dr.delivery_times,
                       c.name as card_name, c.type as card_type, c.api_config,
                       c.text_content, c.data_content, c.enabled as card_enabled,
                       c.description as card_description, c.delay_seconds as card_delay_seconds,
                       c.is_multi_spec, c.spec_name, c.spec_value
                FROM delivery_rules dr
                LEFT JOIN cards c ON dr.card_id = c.id
                WHERE dr.enabled = 1 AND c.enabled = 1
                AND (%s LIKE CONCAT('%%', dr.keyword, '%%') OR dr.keyword LIKE CONCAT('%%', %s, '%%'))
                AND (c.is_multi_spec = 0 OR c.is_multi_spec IS NULL)
                ORDER BY
                    CASE
                        WHEN %s LIKE CONCAT('%%', dr.keyword, '%%') THEN LENGTH(dr.keyword)
                        ELSE LENGTH(dr.keyword) / 2
                    END DESC,
                    dr.delivery_times ASC
                ''', (keyword, keyword, keyword))

                rules = []
                for row in cursor.fetchall():
                    # 解析api_config JSON字符串
                    api_config = row[9]
                    if api_config:
                        try:
                            import json
                            api_config = json.loads(api_config)
                        except (json.JSONDecodeError, TypeError):
                            # 如果解析失败，保持原始字符串
                            pass

                    rules.append({
                        'id': row[0],
                        'keyword': row[1],
                        'card_id': row[2],
                        'delivery_count': row[3],
                        'enabled': bool(row[4]),
                        'description': row[5],
                        'delivery_times': row[6] or 0,
                        'card_name': row[7],
                        'card_type': row[8],
                        'api_config': api_config,
                        'text_content': row[10],
                        'data_content': row[11],
                        'card_enabled': bool(row[12]),
                        'card_description': row[13],
                        'card_delay_seconds': row[14] or 0,
                        'is_multi_spec': bool(row[15]) if row[15] is not None else False,
                        'spec_name': row[16],
                        'spec_value': row[17]
                    })

                if rules:
                    logger.info(f"找到兜底匹配规则: {keyword}")
                else:
                    logger.info(f"未找到匹配规则: {keyword}")

                return rules

            except Exception as e:
                logger.error(f"获取发货规则失败: {e}")
                return []

    def get_delivery_rules_by_item_id(self, item_id: str, user_id: int = None) -> List[Dict[str, Any]]:
        """获取指定商品ID的发货规则（别名方法）"""
        return self.get_delivery_rules_by_item(item_id, user_id)

    def get_all_delivery_rules(self, user_id: int = None) -> List[Dict[str, Any]]:
        """获取所有发货规则（别名方法）"""
        return self.get_delivery_rules(user_id)

    # ==================== 角色权限管理方法 ====================
    
    def get_all_roles(self) -> List[Dict[str, Any]]:
        """获取所有角色"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT id, name, description, is_system, created_at, updated_at
                FROM roles
                ORDER BY is_system DESC, created_at DESC
                ''')
                
                roles = []
                for row in cursor.fetchall():
                    roles.append({
                        'id': row[0],
                        'name': row[1],
                        'description': row[2],
                        'is_system': bool(row[3]),
                        'created_at': row[4],
                        'updated_at': row[5]
                    })
                
                return roles
            except Exception as e:
                logger.error(f"获取角色列表失败: {e}")
                return []
    
    def get_role_by_id(self, role_id: int) -> Optional[Dict[str, Any]]:
        """根据ID获取角色信息"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT id, name, description, is_system, created_at, updated_at
                FROM roles WHERE id = %s
                ''', (role_id,))
                result = cursor.fetchone()
                
                if result:
                    return {
                        'id': result[0],
                        'name': result[1],
                        'description': result[2],
                        'is_system': bool(result[3]),
                        'created_at': result[4],
                        'updated_at': result[5]
                    }
                return None
            except Exception as e:
                logger.error(f"获取角色信息失败: {e}")
                return None
    
    def get_user_roles(self, user_id: int) -> List[Dict[str, Any]]:
        """获取用户的所有角色"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT r.id, r.name, r.description, ur.created_at
                FROM roles r
                JOIN user_roles ur ON r.id = ur.role_id
                WHERE ur.user_id = %s
                ORDER BY r.name
                ''', (user_id,))
                
                roles = []
                for row in cursor.fetchall():
                    roles.append({
                        'id': row[0],
                        'name': row[1],
                        'description': row[2],
                        'assigned_at': row[3]
                    })
                
                return roles
            except Exception as e:
                logger.error(f"获取用户角色失败: {e}")
                return []
    
    def get_user_permissions(self, user_id: int) -> List[str]:
        """获取用户的所有权限"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT DISTINCT p.name
                FROM system_permissions p
                JOIN role_permissions rp ON p.id = rp.permission_id
                JOIN user_roles ur ON rp.role_id = ur.role_id
                WHERE ur.user_id = %s
                ''', (user_id,))
                
                return [row[0] for row in cursor.fetchall()]
            except Exception as e:
                logger.error(f"获取用户权限失败: {e}")
                return []
    
    def get_user_menus(self, user_id: int) -> List[Dict[str, Any]]:
        """获取用户可见的菜单"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT DISTINCT m.id, m.name, m.menu_key, m.icon, m.url, m.parent_id, m.sort_order
                FROM menus m
                JOIN role_menus rm ON m.id = rm.menu_id
                JOIN user_roles ur ON rm.role_id = ur.role_id
                WHERE ur.user_id = %s
                ORDER BY m.sort_order, m.name
                ''', (user_id,))
                
                menus = []
                for row in cursor.fetchall():
                    menus.append({
                        'id': row[0],
                        'name': row[1],
                        'menu_key': row[2],
                        'icon': row[3],
                        'url': row[4],
                        'parent_id': row[5],
                        'sort_order': row[6]
                    })
                
                return menus
            except Exception as e:
                logger.error(f"获取用户菜单失败: {e}")
                return []
    
    def assign_role_to_user(self, user_id: int, role_id: int) -> bool:
        """为用户分配角色"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                INSERT IGNORE INTO user_roles (user_id, role_id) 
                VALUES (%s, %s)
                ''', (user_id, role_id))
                
                if cursor.rowcount > 0:
                    self.conn.commit()
                    logger.info(f"角色分配成功: user_id={user_id}, role_id={role_id}")
                    return True
                else:
                    logger.info(f"角色已存在: user_id={user_id}, role_id={role_id}")
                    return True  # 已存在也算成功
            except Exception as e:
                logger.error(f"分配角色失败: {e}")
                self.conn.rollback()
                return False
    
    def remove_role_from_user(self, user_id: int, role_id: int) -> bool:
        """移除用户的角色"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                DELETE FROM user_roles 
                WHERE user_id = %s AND role_id = %s
                ''', (user_id, role_id))
                
                if cursor.rowcount > 0:
                    self.conn.commit()
                    logger.info(f"角色移除成功: user_id={user_id}, role_id={role_id}")
                    return True
                else:
                    logger.warning(f"角色关联不存在: user_id={user_id}, role_id={role_id}")
                    return False
            except Exception as e:
                logger.error(f"移除角色失败: {e}")
                self.conn.rollback()
                return False
    
    def get_role_permissions(self, role_id: int) -> List[Dict[str, Any]]:
        """获取角色的权限"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT p.id, p.name, p.description, p.category
                FROM system_permissions p
                JOIN role_permissions rp ON p.id = rp.permission_id
                WHERE rp.role_id = %s
                ORDER BY p.category, p.name
                ''', (role_id,))
                
                permissions = []
                for row in cursor.fetchall():
                    permissions.append({
                        'id': row[0],
                        'name': row[1],
                        'description': row[2],
                        'category': row[3]
                    })
                
                return permissions
            except Exception as e:
                logger.error(f"获取角色权限失败: {e}")
                return []
    
    def get_role_menus(self, role_id: int) -> List[Dict[str, Any]]:
        """获取角色的菜单"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT m.id, m.name, m.menu_key, m.icon, m.url, m.parent_id, m.sort_order
                FROM menus m
                JOIN role_menus rm ON m.id = rm.menu_id
                WHERE rm.role_id = %s
                ORDER BY m.sort_order, m.name
                ''', (role_id,))
                
                menus = []
                for row in cursor.fetchall():
                    menus.append({
                        'id': row[0],
                        'name': row[1],
                        'menu_key': row[2],
                        'icon': row[3],
                        'url': row[4],
                        'parent_id': row[5],
                        'sort_order': row[6]
                    })
                
                return menus
            except Exception as e:
                logger.error(f"获取角色菜单失败: {e}")
                return []
    
    def update_role_permissions(self, role_id: int, permission_ids: List[int]) -> bool:
        """更新角色权限"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 删除现有权限
                cursor.execute('DELETE FROM role_permissions WHERE role_id = %s', (role_id,))
                
                # 添加新权限
                for permission_id in permission_ids:
                    cursor.execute('''
                    INSERT INTO role_permissions (role_id, permission_id) 
                    VALUES (%s, %s)
                    ''', (role_id, permission_id))
                
                self.conn.commit()
                logger.info(f"角色权限更新成功: role_id={role_id}")
                return True
            except Exception as e:
                logger.error(f"更新角色权限失败: {e}")
                self.conn.rollback()
                return False
    
    def update_role_menus(self, role_id: int, menu_ids: List[int]) -> bool:
        """更新角色菜单"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 删除现有菜单
                cursor.execute('DELETE FROM role_menus WHERE role_id = %s', (role_id,))
                
                # 添加新菜单
                for menu_id in menu_ids:
                    cursor.execute('''
                    INSERT INTO role_menus (role_id, menu_id) 
                    VALUES (%s, %s)
                    ''', (role_id, menu_id))
                
                self.conn.commit()
                logger.info(f"角色菜单更新成功: role_id={role_id}")
                return True
            except Exception as e:
                logger.error(f"更新角色菜单失败: {e}")
                self.conn.rollback()
                return False
    
    def get_all_permissions(self) -> List[Dict[str, Any]]:
        """获取所有权限"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT id, name, description, category, created_at
                FROM system_permissions
                ORDER BY category, name
                ''')
                
                permissions = []
                for row in cursor.fetchall():
                    permissions.append({
                        'id': row[0],
                        'name': row[1],
                        'description': row[2],
                        'category': row[3],
                        'created_at': row[4]
                    })
                
                return permissions
            except Exception as e:
                logger.error(f"获取权限列表失败: {e}")
                return []
    
    def get_all_menus(self) -> List[Dict[str, Any]]:
        """获取所有菜单"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT id, name, menu_key, icon, url, parent_id, sort_order, is_system, created_at
                FROM menus
                ORDER BY sort_order, name
                ''')
                
                menus = []
                for row in cursor.fetchall():
                    menus.append({
                        'id': row[0],
                        'name': row[1],
                        'menu_key': row[2],
                        'icon': row[3],
                        'url': row[4],
                        'parent_id': row[5],
                        'sort_order': row[6],
                        'is_system': bool(row[7]),
                        'created_at': row[8]
                    })
                
                return menus
            except Exception as e:
                logger.error(f"获取菜单列表失败: {e}")
                return []

    # -------------------- 拉黑记录管理 --------------------
    def add_blacklist_record(self, cookie_id: str, session_id: str, user_id: str = None, 
                           user_name: str = None, reason: str = "主动拉黑") -> bool:
        """添加拉黑记录"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                INSERT INTO blacklist_records (cookie_id, session_id, user_id, user_name, reason, created_at)
                VALUES (%s, %s, %s, %s, %s, CURRENT_TIMESTAMP)
                ''', (cookie_id, session_id, user_id, user_name, reason))
                
                self.conn.commit()
                logger.info(f"拉黑记录添加成功: cookie_id={cookie_id}, session_id={session_id}")
                return True
            except Exception as e:
                logger.error(f"添加拉黑记录失败: {e}")
                self.conn.rollback()
                return False

    def get_blacklist_records(self, cookie_id: str = None) -> List[Dict[str, Any]]:
        """获取拉黑记录列表"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                if cookie_id:
                    cursor.execute('''
                    SELECT id, cookie_id, session_id, user_id, user_name, reason, created_at
                    FROM blacklist_records
                    WHERE cookie_id = %s
                    ORDER BY created_at DESC
                    ''', (cookie_id,))
                else:
                    cursor.execute('''
                    SELECT id, cookie_id, session_id, user_id, user_name, reason, created_at
                    FROM blacklist_records
                    ORDER BY created_at DESC
                    ''')
                
                records = []
                for row in cursor.fetchall():
                    records.append({
                        'id': row[0],
                        'cookie_id': row[1],
                        'session_id': row[2],
                        'user_id': row[3],
                        'user_name': row[4],
                        'reason': row[5],
                        'created_at': row[6]
                    })
                return records
            except Exception as e:
                logger.error(f"获取拉黑记录失败: {e}")
                return []

    def is_user_blacklisted(self, cookie_id: str, session_id: str) -> bool:
        """检查用户是否已被拉黑"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT COUNT(*) FROM blacklist_records
                WHERE cookie_id = %s AND session_id = %s
                ''', (cookie_id, session_id))
                
                count = cursor.fetchone()[0]
                return count > 0
            except Exception as e:
                logger.error(f"检查拉黑状态失败: {e}")
                return False

    def remove_blacklist_record(self, cookie_id: str, session_id: str) -> bool:
        """移除拉黑记录"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                DELETE FROM blacklist_records
                WHERE cookie_id = %s AND session_id = %s
                ''', (cookie_id, session_id))
                
                if cursor.rowcount > 0:
                    self.conn.commit()
                    logger.info(f"拉黑记录移除成功: cookie_id={cookie_id}, session_id={session_id}")
                    return True
                else:
                    logger.warning(f"未找到要移除的拉黑记录: cookie_id={cookie_id}, session_id={session_id}")
                    return False
            except Exception as e:
                logger.error(f"移除拉黑记录失败: {e}")
                self.conn.rollback()
                return False

    # -------------------- 聚合豆管理操作 --------------------
    
    def get_user_balance(self, user_id: int) -> float:
        """获取用户聚合豆余额"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                self._execute_sql(cursor, "SELECT balance FROM user_balance WHERE user_id = %s", (user_id,))
                result = cursor.fetchone()
                return float(result[0]) if result else 0.0
            except Exception as e:
                logger.error(f"获取用户余额失败: {e}")
                return 0.0

    def init_user_balance(self, user_id: int) -> bool:
        """初始化用户聚合豆余额"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                self._execute_sql(cursor, '''
                INSERT IGNORE INTO user_balance (user_id, balance)
                VALUES (%s, 0.00)
                ''', (user_id,))
                self.conn.commit()
                logger.info(f"用户余额初始化成功: user_id={user_id}")
                return True
            except Exception as e:
                logger.error(f"初始化用户余额失败: {e}")
                self.conn.rollback()
                return False

    def update_user_balance(self, user_id: int, amount: float, order_id: str = None, 
                           record_type: str = 'consume', description: str = '', 
                           admin_user_id: int = None) -> bool:
        """更新用户聚合豆余额并记录变动"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 开始事务
                cursor.execute("START TRANSACTION")
                
                # 获取当前余额
                current_balance = self.get_user_balance(user_id)
                
                # 计算新余额
                new_balance = current_balance + amount
                
                # 检查余额是否足够（消费时）
                if amount < 0 and new_balance < 0:
                    cursor.execute("ROLLBACK")
                    logger.warning(f"用户余额不足: user_id={user_id}, 当前余额={current_balance}, 尝试消费={abs(amount)}")
                    return False
                
                # 更新余额
                self._execute_sql(cursor, '''
                INSERT INTO user_balance (user_id, balance)
                VALUES (%s, %s)
                ON DUPLICATE KEY UPDATE
                balance = %s,
                updated_at = CURRENT_TIMESTAMP
                ''', (user_id, new_balance, new_balance))
                
                # 记录变动
                self._execute_sql(cursor, '''
                INSERT INTO balance_records (user_id, order_id, amount, balance_after, record_type, description, admin_user_id)
                VALUES (%s, %s, %s, %s, %s, %s, %s)
                ''', (user_id, order_id, amount, new_balance, record_type, description, admin_user_id))
                
                self.conn.commit()
                logger.info(f"用户余额更新成功: user_id={user_id}, 变动={amount}, 新余额={new_balance}, 类型={record_type}")
                return True
            except Exception as e:
                logger.error(f"更新用户余额失败: {e}")
                self.conn.rollback()
                return False

    def get_balance_records(self, user_id: int, page: int = 1, page_size: int = 20) -> Dict[str, Any]:
        """获取用户聚合豆消费记录"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 计算偏移量
                offset = (page - 1) * page_size
                
                # 获取总记录数
                self._execute_sql(cursor, "SELECT COUNT(*) FROM balance_records WHERE user_id = %s", (user_id,))
                total_count = cursor.fetchone()[0]
                
                # 获取分页记录
                self._execute_sql(cursor, '''
                SELECT br.id, br.order_id, br.amount, br.balance_after, br.record_type, 
                       br.description, br.created_at, u.username as admin_username
                FROM balance_records br
                LEFT JOIN users u ON br.admin_user_id = u.id
                WHERE br.user_id = %s
                ORDER BY br.created_at DESC
                LIMIT %s OFFSET %s
                ''', (user_id, page_size, offset))
                
                records = []
                for row in cursor.fetchall():
                    records.append({
                        'id': row[0],
                        'order_id': row[1],
                        'amount': float(row[2]),
                        'balance_after': float(row[3]),
                        'record_type': row[4],  # 前端期望的字段名是record_type
                        'description': row[5],
                        'created_at': row[6],
                        'admin_username': row[7]
                    })
                
                # 计算分页信息
                total_pages = (total_count + page_size - 1) // page_size
                
                return {
                    'records': records,
                    'pagination': {
                        'current_page': page,
                        'page_size': page_size,
                        'total_count': total_count,
                        'total_pages': total_pages,
                        'has_next': page < total_pages,
                        'has_prev': page > 1
                    }
                }
            except Exception as e:
                logger.error(f"获取余额记录失败: {e}")
                return {
                    'records': [],
                    'pagination': {
                        'current_page': page,
                        'page_size': page_size,
                        'total_count': 0,
                        'total_pages': 0,
                        'has_next': False,
                        'has_prev': False
                    }
                }

    def admin_adjust_balance(self, user_id: int, amount: float, description: str, admin_user_id: int) -> bool:
        """管理员调整用户聚合豆余额"""
        record_type = 'admin_adjust'
        if amount > 0:
            description = f"管理员充值: {description}" if description else "管理员充值"
        else:
            description = f"管理员扣除: {description}" if description else "管理员扣除"
        
        return self.update_user_balance(
            user_id=user_id,
            amount=amount,
            record_type=record_type,
            description=description,
            admin_user_id=admin_user_id
        )

    def consume_balance_for_order(self, user_id: int, order_id: str, amount: float, description: str = '') -> bool:
        """为订单消费聚合豆"""
        if not description:
            description = f"订单消费: {order_id}"
        
        return self.update_user_balance(
            user_id=user_id,
            amount=-abs(amount),  # 确保是负数
            order_id=order_id,
            record_type='consume',
            description=description
        )

    def add_balance_record(self, user_id: int, amount: float, operation_type: str, 
                          reason: str = '', order_id: str = None, admin_user_id: int = None) -> bool:
        """添加聚合豆记录（不更新余额，仅记录）"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 获取当前余额
                current_balance = self.get_user_balance(user_id)
                
                # 插入记录
                self._execute_sql(cursor, '''
                INSERT INTO balance_records (user_id, order_id, amount, balance_after, record_type, description, admin_user_id)
                VALUES (%s, %s, %s, %s, %s, %s, %s)
                ''', (user_id, order_id, amount, current_balance, operation_type, reason, admin_user_id))
                
                self.conn.commit()
                logger.info(f"聚合豆记录添加成功: user_id={user_id}, amount={amount}, type={operation_type}")
                return True
            except Exception as e:
                logger.error(f"添加聚合豆记录失败: {e}")
                self.conn.rollback()
                return False

    def get_user_id_by_cookie(self, cookie_id: str) -> Optional[int]:
        """根据cookie_id获取对应的用户ID"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                self._execute_sql(cursor, "SELECT user_id FROM cookies WHERE id = %s", (cookie_id,))
                result = cursor.fetchone()
                return result[0] if result else None
            except Exception as e:
                logger.error(f"根据cookie_id获取用户ID失败: {e}")
                return None

    def check_and_consume_balance(self, user_id: int, amount: float, description: str = "", order_id: str = None) -> bool:
        """检查用户余额并扣费，如果余额不足返回False"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                
                # 开始事务
                cursor.execute("START TRANSACTION")
                
                # 获取当前余额
                current_balance = self.get_user_balance(user_id)
                
                # 检查余额是否足够
                if current_balance < amount:
                    cursor.execute("ROLLBACK")
                    logger.warning(f"用户余额不足: user_id={user_id}, 当前余额={current_balance}, 需要消费={amount}")
                    return False
                
                # 计算新余额
                new_balance = current_balance - amount
                
                # 更新余额
                self._execute_sql(cursor, '''
                INSERT INTO user_balance (user_id, balance)
                VALUES (%s, %s)
                ON DUPLICATE KEY UPDATE
                balance = %s,
                updated_at = CURRENT_TIMESTAMP
                ''', (user_id, new_balance, new_balance))
                
                # 记录变动
                self._execute_sql(cursor, '''
                INSERT INTO balance_records (user_id, order_id, amount, balance_after, record_type, description)
                VALUES (%s, %s, %s, %s, %s, %s)
                ''', (user_id, order_id, -amount, new_balance, 'consume', description))
                
                self.conn.commit()
                logger.info(f"用户余额扣费成功: user_id={user_id}, 扣费={amount}, 新余额={new_balance}, 描述={description}")
                return True
            except Exception as e:
                logger.error(f"用户余额扣费失败: {e}")
                self.conn.rollback()
                return False

    def add_pending_payment_check(self, cookie_id: str, order_id: str, user_id: str, item_id: str, delay_seconds: int = 30) -> int:
        """添加待付款订单的定时检查任务"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                import datetime
                scheduled_time = datetime.datetime.now() + datetime.timedelta(seconds=delay_seconds)
                trigger_time = datetime.datetime.now()
                
                self._execute_sql(cursor, '''
                INSERT INTO pending_payment_checks 
                (cookie_id, order_id, user_id, item_id, trigger_time, scheduled_time, status)
                VALUES (%s, %s, %s, %s, %s, %s, 'pending')
                ON DUPLICATE KEY UPDATE
                trigger_time = VALUES(trigger_time),
                scheduled_time = VALUES(scheduled_time),
                status = 'pending',
                checked_at = NULL,
                cancelled_at = NULL
                ''', (cookie_id, order_id, user_id, item_id, trigger_time, scheduled_time))
                
                task_id = cursor.lastrowid
                self.conn.commit()
                logger.info(f"添加待付款检查任务: {task_id}, 订单: {order_id}, 账号: {cookie_id}, 用户: {user_id}")
                return task_id
                
            except Exception as e:
                logger.error(f"添加待付款检查任务失败: {e}")
                self.conn.rollback()
                return 0

    def cancel_pending_payment_check(self, cookie_id: str, order_id: str) -> bool:
        """取消指定订单的待付款检查任务"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                import datetime
                current_time = datetime.datetime.now()
                
                self._execute_sql(cursor, '''
                UPDATE pending_payment_checks 
                SET status = 'cancelled', cancelled_at = %s
                WHERE cookie_id = %s AND order_id = %s AND status = 'pending'
                ''', (current_time, cookie_id, order_id))
                
                affected_rows = cursor.rowcount
                self.conn.commit()
                if affected_rows > 0:
                    logger.info(f"取消待付款检查任务: 订单:{order_id}, 账号:{cookie_id}")
                
                return affected_rows > 0
                
            except Exception as e:
                logger.error(f"取消待付款检查任务失败: {e}")
                self.conn.rollback()
                return False

    def get_pending_payment_checks(self) -> List[Dict[str, Any]]:
        """获取待处理的付款检查任务"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                cursor.execute('''
                SELECT id, cookie_id, order_id, user_id, item_id, trigger_time, 
                       scheduled_time, status, created_at
                FROM pending_payment_checks 
                WHERE status = 'pending' AND scheduled_time <= NOW()
                ORDER BY scheduled_time ASC
                ''')
                
                tasks = []
                for row in cursor.fetchall():
                    tasks.append({
                        'id': row[0],
                        'cookie_id': row[1],
                        'order_id': row[2],
                        'user_id': row[3],
                        'item_id': row[4],
                        'trigger_time': row[5],
                        'scheduled_time': row[6],
                        'status': row[7],
                        'created_at': row[8]
                    })
                
                return tasks
            except Exception as e:
                logger.error(f"获取待处理付款检查任务失败: {e}")
                return []

    def mark_pending_payment_check_completed(self, task_id: int) -> bool:
        """标记待付款检查任务为已完成"""
        with self.lock:
            try:
                cursor = self.get_connection().cursor()
                import datetime
                current_time = datetime.datetime.now()
                
                self._execute_sql(cursor, '''
                UPDATE pending_payment_checks 
                SET status = 'completed', checked_at = %s
                WHERE id = %s
                ''', (current_time, task_id))
                
                self.conn.commit()
                return cursor.rowcount > 0
                
            except Exception as e:
                logger.error(f"标记待付款检查任务为已完成失败: {e}")
                self.conn.rollback()
                return False

# 创建全局数据库管理器实例
db_manager = DBManager()